From fc1b85113d7cd2a1688af09cbb92be2fb3c15a2f Mon Sep 17 00:00:00 2001
From: jillion <439912596@qq.com>
Date: Mon, 21 Sep 2026 18:22:52 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BA=8B=E4=BB=B6=E5=BC=82=E5=B8=B8=20?=
 =?UTF-8?q?Worker=20=E5=B7=A1=E6=A3=80=E5=91=8A=E8=AD=A6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 inquiry-agent/agent/jobs/__init__.py          |  17 ++
 .../agent/jobs/event_exception_poll.py        | 268 ++++++++++++++++++
 inquiry-agent/tests/test_event_exception.py   |  58 ++++
 3 files changed, 343 insertions(+)
 create mode 100644 inquiry-agent/agent/jobs/event_exception_poll.py

diff --git a/inquiry-agent/agent/jobs/__init__.py b/inquiry-agent/agent/jobs/__init__.py
index 8eced99..3737806 100644
--- a/inquiry-agent/agent/jobs/__init__.py
+++ b/inquiry-agent/agent/jobs/__init__.py
@@ -112,6 +112,7 @@ class WorkerRuntime:
             threading.Thread(target=self._archive_loop, name="archive-loop", daemon=True),
             threading.Thread(target=self._wshoto_poll_loop, name="wshoto-poll", daemon=True),
             threading.Thread(target=self._deal_followup_loop, name="deal-followup", daemon=True),
+            threading.Thread(target=self._event_exception_loop, name="event-exception", daemon=True),
             threading.Thread(target=self._pdf_loop, name="pdf-loop", daemon=True),
         ]
         for t in self._threads:
@@ -290,6 +291,22 @@ class WorkerRuntime:
                 logger.exception("协商再问轮询失败")
             self._stop.wait(60.0)
 
+    def _event_exception_loop(self) -> None:
+        """
+        HTTP 槽巡检事件异常。独立线程，不占回调、不与 soffice 串队。
+        """
+        from agent.jobs.event_exception_poll import poll_event_exceptions
+
+        while not self._stop.is_set():
+            try:
+                self.http_jobs.run(
+                    lambda: poll_event_exceptions(settings=self.settings),
+                    label="event-exception",
+                )
+            except Exception:
+                logger.exception("事件异常巡检失败")
+            self._stop.wait(60.0)
+
     def _pdf_loop(self) -> None:
         """仅 LibreOffice 单槽消费 PDF Stream。"""
         if not self.redis:
diff --git a/inquiry-agent/agent/jobs/event_exception_poll.py b/inquiry-agent/agent/jobs/event_exception_poll.py
new file mode 100644
index 0000000..53fd5a6
--- /dev/null
+++ b/inquiry-agent/agent/jobs/event_exception_poll.py
@@ -0,0 +1,268 @@
+"""
+事件异常巡检：Worker HTTP 槽轮询到点工单并告警。
+
+本文件职责：读主账等待、按工作日对照超时小时、会话喊人、可选应用卡片、再记摘要。
+禁止：回调线程调用；占 LibreOffice 槽；改六态；反复催已告警的等待。
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from typing import Any, Callable, Optional
+
+from agent.config import Settings, get_settings
+from agent.policy import inquiry_copy as copy
+from agent.policy.deal_followup import SHANGHAI, workday_elapsed_hours
+from agent.policy.event_exception import (
+    card_title,
+    conversation_of,
+    exception_type,
+    pick_app_targets,
+    pick_mentions,
+    timeout_code,
+)
+
+logger = logging.getLogger(__name__)
+
+SendGroupFn = Callable[..., bool]
+EnqueueFn = Callable[..., tuple[bool, str]]
+TimeoutLookupFn = Callable[..., dict[str, Any]]
+
+
+def _parse_started(raw: Any) -> Optional[datetime]:
+    text = str(raw or "").strip()
+    if not text:
+        return None
+    text = text.replace("T", " ")[:19]
+    try:
+        naive = datetime.strptime(text, "%Y-%m-%d %H:%M:%S")
+        return naive.replace(tzinfo=SHANGHAI)
+    except ValueError:
+        return None
+
+
+def _hours_of(row: dict[str, Any]) -> float:
+    try:
+        return float(row.get("timeoutHours") if row.get("timeoutHours") is not None else row.get("timeout_hours") or 0)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def poll_event_exceptions(
+    *,
+    ledger: Any = None,
+    settings: Optional[Settings] = None,
+    now: Optional[datetime] = None,
+    enqueue: Optional[EnqueueFn] = None,
+    send_group: Optional[SendGroupFn] = None,
+    timeout_lookup: Optional[TimeoutLookupFn] = None,
+    notify_ready: Optional[bool] = None,
+) -> int:
+    """
+    巡检未告警等待。返回本轮成功记下异常的工单数。
+
+    会话成功才 record；卡片失败不影响会话。禁止进企微回调线程。
+    """
+    cfg = settings or get_settings()
+    book = ledger
+    if book is None:
+        if str(getattr(cfg, "ledger_backend", "http") or "http").lower() == "memory":
+            return 0
+        from agent.ledger.http_ledger import HttpLedger
+
+        book = HttpLedger()
+    lister = getattr(book, "list_event_waits", None)
+    if not callable(lister):
+        return 0
+    stamp = now or datetime.now(SHANGHAI)
+    if stamp.tzinfo is None:
+        stamp = stamp.replace(tzinfo=SHANGHAI)
+    test_prefix = str(getattr(cfg, "ytd_env", "") or "").strip().lower() == "test"
+    lookup = timeout_lookup or getattr(book, "get_timeout_by_code", None)
+    ready = notify_app_flag(notify_ready)
+    sent = 0
+    for row in lister() or []:
+        if not isinstance(row, dict):
+            continue
+        try:
+            if _handle_one(
+                row,
+                book=book,
+                now=stamp,
+                test_prefix=test_prefix,
+                enqueue=enqueue,
+                send_group=send_group,
+                timeout_lookup=lookup,
+                notify_ready=ready,
+                settings=cfg,
+            ):
+                sent += 1
+        except Exception:
+            logger.exception(
+                "事件异常巡检失败 wo=%s",
+                row.get("work_order_no") or row.get("workOrderNo"),
+            )
+    return sent
+
+
+def notify_app_flag(override: Optional[bool]) -> bool:
+    if override is not None:
+        return bool(override)
+    from agent.policy.deal_outcome import notify_app_ready
+
+    return notify_app_ready()
+
+
+def _handle_one(
+    row: dict[str, Any],
+    *,
+    book: Any,
+    now: datetime,
+    test_prefix: bool,
+    enqueue: Optional[EnqueueFn],
+    send_group: Optional[SendGroupFn],
+    timeout_lookup: Any,
+    notify_ready: bool,
+    settings: Settings,
+) -> bool:
+    no = str(row.get("work_order_no") or row.get("workOrderNo") or "").strip()
+    kind = str(row.get("event_wait_kind") or row.get("eventWaitKind") or "").strip()
+    if not no or not kind:
+        return False
+    line = str(row.get("business_line") or row.get("businessLine") or "").strip()
+    facts = dict(row.get("facts") or {})
+    klass = str(facts.get("运输分类") or facts.get("transportType") or "").strip()
+    code = timeout_code(wait_kind=kind, business_line=line, transport_class=klass)
+    if not code or not callable(timeout_lookup):
+        return False
+    timeout_row = timeout_lookup(timeout_code=code) or {}
+    if not timeout_row.get("found"):
+        return False
+    hours = _hours_of(timeout_row)
+    if hours <= 0:
+        return False
+    started = _parse_started(row.get("event_wait_started_at") or row.get("eventWaitStartedAt"))
+    if started is None:
+        return False
+    if workday_elapsed_hours(started, now) < hours:
+        return False
+    watchers = [dict(x) for x in (row.get("watchers") or []) if isinstance(x, dict)]
+    sales = next((dict(w) for w in watchers if str(w.get("roleCode") or "").lower() in {"sales", ""}), None)
+    if sales is None:
+        sales = {
+            "wecomId": str(row.get("sales_wecom_id") or row.get("salesWecomId") or "").strip(),
+            "name": str(row.get("sales_name") or row.get("salesName") or "销售").strip() or "销售",
+            "roleCode": "sales",
+            "notifyEventException": 0,
+        }
+    people = pick_mentions(wait_kind=kind, sales=sales, members=watchers, business_line=line)
+    names = [str(p.get("name") or "").strip() for p in people if str(p.get("name") or "").strip()]
+    chat_id = str(row.get("collab_chat_id") or row.get("collabChatId") or "").strip()
+    where = conversation_of(wait_kind=kind, collab_chat_id=chat_id)
+    mention = where == "group"
+    text = copy.event_exception_session_text(
+        exception_type=exception_type(kind),
+        work_order_no=no,
+        names=names,
+        hours=hours,
+        mention=mention,
+        test_prefix=test_prefix,
+    )
+    ok_session = False
+    if where == "group":
+        if not chat_id:
+            logger.warning("事件异常无协同群 wo=%s kind=%s", no, kind)
+            return False
+        sender = send_group or _default_send_group
+        ok_session = bool(sender(chat_id=chat_id, content=text, business_line=line))
+    else:
+        uid = str((people[0].get("wecomId") if people else "") or sales.get("wecomId") or "").strip()
+        if not uid:
+            logger.warning("事件异常无私聊对象 wo=%s kind=%s", no, kind)
+            return False
+        enq = enqueue or _default_enqueue
+        ok_session, info = enq(
+            touser=uid,
+            content=text,
+            dedupe_key=f"event-ex:{no}:{kind}",
+            payload={},
+        )
+        if not ok_session:
+            logger.warning("事件异常私聊入队失败 wo=%s info=%s", no, info)
+    if not ok_session:
+        return False
+    if notify_ready:
+        _send_app_cards(
+            people=people,
+            kind=kind,
+            work_order_no=no,
+            hours=hours,
+            test_prefix=test_prefix,
+            enqueue=enqueue or _default_enqueue,
+        )
+    recorder = getattr(book, "record_event_exception", None)
+    if callable(recorder):
+        recorder(work_order_no=no, event_exception=exception_type(kind))
+    return True
+
+
+def _send_app_cards(
+    *,
+    people: list[dict[str, Any]],
+    kind: str,
+    work_order_no: str,
+    hours: float,
+    test_prefix: bool,
+    enqueue: EnqueueFn,
+) -> None:
+    payload = copy.event_exception_alert_payload(
+        card_title=card_title(kind),
+        work_order_no=work_order_no,
+        hours=hours,
+        test_prefix=test_prefix,
+    )
+    fallback = copy.event_exception_alert_text(
+        card_title=card_title(kind),
+        work_order_no=work_order_no,
+        hours=hours,
+        test_prefix=test_prefix,
+    )
+    for person in pick_app_targets(people):
+        uid = str(person.get("wecomId") or "").strip()
+        if not uid:
+            continue
+        ok, info = enqueue(
+            touser=uid,
+            content=fallback,
+            dedupe_key=f"event-ex-app:{work_order_no}:{kind}:{uid}",
+            payload=payload,
+        )
+        if not ok:
+            logger.warning("事件异常应用卡片入队失败 uid=%s info=%s", uid, info)
+
+
+def _default_enqueue(**kwargs: Any) -> tuple[bool, str]:
+    from agent.channel.queue import get_message_store
+
+    return get_message_store().enqueue_outbound(**kwargs)
+
+
+def _default_send_group(*, chat_id: str, content: str, business_line: str = "") -> bool:
+    line = (business_line or "").strip().upper()
+    if line == "AIR":
+        from agent.channel.aibot.reply import AibotReplyClient
+
+        out = AibotReplyClient().send_group(chat_id=chat_id, content=content)
+        return bool(out.get("ok"))
+    from agent.channel.outbox.sender import WeComAppClient
+    from agent.config import get_settings
+
+    cfg = get_settings()
+    client = WeComAppClient(
+        corp_id=cfg.wecom_corp_id,
+        secret=cfg.wecom_secret,
+        agent_id=int(cfg.wecom_agent_id or "0") or 0,
+    )
+    out = client.send_group(chat_id=chat_id, content=content)
+    return bool(out.get("ok"))
diff --git a/inquiry-agent/tests/test_event_exception.py b/inquiry-agent/tests/test_event_exception.py
index 5834154..a90c18e 100644
--- a/inquiry-agent/tests/test_event_exception.py
+++ b/inquiry-agent/tests/test_event_exception.py
@@ -190,5 +190,63 @@ class TestEventExceptionCopy(unittest.TestCase):
         self.assertIn("您已超过 48 工作小时无响应", card["sub_title_text"])
 
 
+class TestPoll(unittest.TestCase):
+    def test_fires_once_on_timeout(self) -> None:
+        import importlib.util
+        from datetime import datetime, timedelta
+
+        from agent.ledger.memory_ledger import MemoryLedger
+        from agent.policy.deal_followup import SHANGHAI
+
+        path = os.path.join(ROOT, "agent", "jobs", "event_exception_poll.py")
+        spec = importlib.util.spec_from_file_location("ee_poll", path)
+        self.assertIsNotNone(spec)
+        self.assertIsNotNone(spec.loader)
+        mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+        poll_event_exceptions = mod.poll_event_exceptions
+
+        book = MemoryLedger()
+        book.set_sea_staff(
+            [
+                {
+                    "wecomId": "p1",
+                    "name": "张三",
+                    "roleCode": "sea",
+                    "notifyEventException": 1,
+                    "status": "active",
+                }
+            ]
+        )
+        created = book.create_ticket(sender_id="s1", business_line="SEA", facts={"运输分类": "其他"})
+        wo = created["work_order_no"]
+        book.start_event_wait(work_order_no=wo, wait_kind="product_quote")
+        t = book.get_ticket(work_order_no=wo)
+        t.collab_chat_id = "wr1"
+        t.collab_member_ids = ["p1"]
+        t.event_wait_started_at = (datetime.now(SHANGHAI) - timedelta(hours=3)).strftime(
+            "%Y-%m-%d %H:%M:%S"
+        )
+        book._timeouts["sea_product_other"] = 0.25
+        groups: list[dict] = []
+
+        def send_group(*, chat_id: str, content: str, business_line: str = "") -> bool:
+            groups.append({"chat_id": chat_id, "content": content, "line": business_line})
+            return True
+
+        now = datetime.now(SHANGHAI)
+        n = poll_event_exceptions(
+            ledger=book, now=now, send_group=send_group, notify_ready=False
+        )
+        self.assertEqual(n, 1)
+        self.assertEqual(book.get_ticket(work_order_no=wo).event_exception, "产品响应超时")
+        self.assertTrue(groups)
+        self.assertIn("@张三", groups[0]["content"])
+        n2 = poll_event_exceptions(
+            ledger=book, now=now, send_group=send_group, notify_ready=False
+        )
+        self.assertEqual(n2, 0)
+
+
 if __name__ == "__main__":
     unittest.main()