feat: 事件异常主账起停等待接口

Co-authored-by: Cursor <cursoragent@cursor.com>

inquiry-agent/agent/ledger/http_ledger.py

@@ -477,3 +477,49 @@ class HttpLedger:
         if isinstance(rows, list):
             return [dict(x) for x in rows if isinstance(x, dict)]
         return []
+
+    def start_event_wait(self, *, work_order_no: str, wait_kind: str) -> dict[str, Any]:
+        out = self._post(
+            "/inquiry/agent/ticket/startEventWait",
+            {"workOrderNo": work_order_no, "waitKind": wait_kind},
+        )
+        if out.get("ok") is False:
+            return out
+        out.setdefault("ok", True)
+        return out
+
+    def clear_event_wait(self, *, work_order_no: str) -> dict[str, Any]:
+        out = self._post(
+            "/inquiry/agent/ticket/clearEventWait",
+            {"workOrderNo": work_order_no},
+        )
+        if out.get("ok") is False:
+            return out
+        out.setdefault("ok", True)
+        return out
+
+    def list_event_waits(self) -> list[dict[str, Any]]:
+        out = self._post("/inquiry/agent/ticket/listEventWaits", {})
+        rows = out.get("result") if isinstance(out.get("result"), list) else None
+        if rows is None:
+            rows = out.get("tickets") if isinstance(out.get("tickets"), list) else []
+        return [dict(x) for x in rows if isinstance(x, dict)]
+
+    def record_event_exception(self, *, work_order_no: str, event_exception: str) -> dict[str, Any]:
+        out = self._post(
+            "/inquiry/agent/ticket/recordEventException",
+            {"workOrderNo": work_order_no, "eventException": event_exception},
+        )
+        if out.get("ok") is False:
+            return out
+        out.setdefault("ok", True)
+        return out
+
+    def get_timeout_by_code(self, *, timeout_code: str) -> dict[str, Any]:
+        out = self._post(
+            "/inquiry/agent/responseTimeout/byCode",
+            {"timeoutCode": timeout_code},
+        )
+        if "found" not in out:
+            return {"found": False, "timeoutCode": timeout_code}
+        return out

inquiry-agent/agent/ledger/memory_ledger.py

@@ -44,6 +44,10 @@ class MemoryTicket:
     lost_reason: str = ""
     next_followup_at: str = ""
     deal_notify_summary: str = ""
+    event_exception: str = ""
+    event_wait_kind: str = ""
+    event_wait_started_at: str = ""
+    event_wait_alerted: int = 0
     system_exception: str = ""
     system_exception_reason: str = ""
     system_exception_step: str = ""
@@ -77,6 +81,7 @@ class MemoryLedger:
         self._staff_by_wecom: dict[str, dict[str, Any]] = {}
         self._staff_by_name: dict[str, dict[str, Any]] = {}
         self._exception_notify_users: list[dict[str, Any]] = []
+        self._timeouts: dict[str, float] = {}
 
     def _next_no(self) -> str:
         self._seq += 1
@@ -727,3 +732,119 @@ class MemoryLedger:
             "releaseId": f"CA-{no}-RLSE",
             "tmsRequestId": f"TMS-RLSE-{no}",
         }
+
+    def start_event_wait(self, *, work_order_no: str, wait_kind: str) -> dict[str, Any]:
+        """开始盯等待。同种类不重置起点。不改六态。"""
+
+        no = (work_order_no or "").strip()
+        kind = (wait_kind or "").strip()
+        with self._lock:
+            ticket = self._tickets.get(no)
+            if ticket is None:
+                return {"ok": False, "error": "not_found"}
+            if ticket.status in {"已成交", "已关闭", "转人工"} or not kind:
+                return {"ok": False, "error": "not_watchable", "status": ticket.status}
+            if ticket.event_wait_kind == kind and ticket.event_wait_started_at:
+                return {"ok": True, "work_order_no": no, "eventWaitKind": kind}
+            ticket.event_wait_kind = kind
+            ticket.event_wait_started_at = datetime.now(_SHANGHAI).strftime("%Y-%m-%d %H:%M:%S")
+            ticket.event_wait_alerted = 0
+            return {"ok": True, "work_order_no": no, "eventWaitKind": kind}
+
+    def clear_event_wait(self, *, work_order_no: str) -> dict[str, Any]:
+        no = (work_order_no or "").strip()
+        with self._lock:
+            ticket = self._tickets.get(no)
+            if ticket is None:
+                return {"ok": False, "error": "not_found"}
+            ticket.event_wait_kind = ""
+            ticket.event_wait_started_at = ""
+            ticket.event_wait_alerted = 0
+            return {"ok": True, "work_order_no": no}
+
+    def list_event_waits(self) -> list[dict[str, Any]]:
+        out: list[dict[str, Any]] = []
+        with self._lock:
+            for ticket in self._tickets.values():
+                if not (ticket.event_wait_kind or "").strip():
+                    continue
+                if int(ticket.event_wait_alerted or 0) != 0:
+                    continue
+                if ticket.status in {"已成交", "已关闭", "转人工"}:
+                    continue
+                watchers: list[dict[str, Any]] = []
+                seen: set[str] = set()
+                sales_id = (ticket.sales_wecom_id or "").strip()
+                if sales_id:
+                    row = dict(self._staff_by_wecom.get(sales_id) or {})
+                    watchers.append(
+                        {
+                            "wecomId": sales_id,
+                            "name": str(row.get("name") or ticket.sales_wecom_id),
+                            "roleCode": str(row.get("roleCode") or row.get("role_code") or "sales"),
+                            "notifyEventException": row.get("notifyEventException", 0),
+                        }
+                    )
+                    seen.add(sales_id)
+                for uid in ticket.collab_member_ids:
+                    key = str(uid).strip()
+                    if not key or key in seen:
+                        continue
+                    row = dict(self._staff_by_wecom.get(key) or {})
+                    if not row:
+                        continue
+                    watchers.append(
+                        {
+                            "wecomId": key,
+                            "name": str(row.get("name") or key),
+                            "roleCode": str(row.get("roleCode") or row.get("role_code") or ""),
+                            "notifyEventException": row.get("notifyEventException", 0),
+                        }
+                    )
+                    seen.add(key)
+                out.append(
+                    {
+                        "work_order_no": ticket.work_order_no,
+                        "workOrderNo": ticket.work_order_no,
+                        "status": ticket.status,
+                        "business_line": ticket.business_line,
+                        "businessLine": ticket.business_line,
+                        "event_wait_kind": ticket.event_wait_kind,
+                        "eventWaitKind": ticket.event_wait_kind,
+                        "event_wait_started_at": ticket.event_wait_started_at,
+                        "eventWaitStartedAt": ticket.event_wait_started_at,
+                        "sales_wecom_id": ticket.sales_wecom_id,
+                        "salesWecomId": ticket.sales_wecom_id,
+                        "sales_name": str(self._staff_by_wecom.get(sales_id, {}).get("name") or "销售"),
+                        "collab_chat_id": ticket.collab_chat_id,
+                        "collabChatId": ticket.collab_chat_id,
+                        "collab_member_ids": list(ticket.collab_member_ids),
+                        "facts": dict(ticket.facts),
+                        "eventException": ticket.event_exception,
+                        "watchers": watchers,
+                    }
+                )
+        return out
+
+    def record_event_exception(self, *, work_order_no: str, event_exception: str) -> dict[str, Any]:
+        from agent.policy.event_exception import append_summary
+
+        no = (work_order_no or "").strip()
+        with self._lock:
+            ticket = self._tickets.get(no)
+            if ticket is None:
+                return {"ok": False, "error": "not_found"}
+            ticket.event_exception = append_summary(ticket.event_exception, event_exception)
+            ticket.event_wait_alerted = 1
+            return {
+                "ok": True,
+                "work_order_no": no,
+                "eventException": ticket.event_exception,
+            }
+
+    def get_timeout_by_code(self, *, timeout_code: str) -> dict[str, Any]:
+        code = (timeout_code or "").strip()
+        hours = self._timeouts.get(code)
+        if hours is None:
+            return {"found": False, "timeoutCode": code}
+        return {"found": True, "timeoutCode": code, "timeoutHours": hours}

inquiry-agent/agent/policy/event_exception.py

@@ -7,11 +7,14 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
 from agent.schema.land_options import canonicalize_land_class
 from agent.schema.sea_options import canonicalize_sea_class
 
+logger = logging.getLogger(__name__)
+
 WAIT_AIR_QUOTE = "air_quote"
 WAIT_PRODUCT_QUOTE = "product_quote"
 WAIT_ADOPT = "adopt"
@@ -106,6 +109,50 @@ def format_timeout_hours(hours: float) -> str:
     return text or "0"
 
 
+def append_summary(existing: str, type_name: str) -> str:
+    """多次异常用分号拼接，同一类型不重复。"""
+    add = (type_name or "").strip()
+    cur = (existing or "").strip()
+    if not add:
+        return cur
+    if not cur or cur == "-":
+        return add
+    parts = [p.strip() for p in cur.split("；") if p.strip()]
+    if add in parts:
+        return cur
+    parts.append(add)
+    return "；".join(parts)
+
+
+def start_wait(ledger: Any, *, work_order_no: str, wait_kind: str) -> None:
+    """开始盯等待。失败只打日志，不抛给回调。"""
+    no = (work_order_no or "").strip()
+    kind = (wait_kind or "").strip()
+    if not no or not kind or ledger is None:
+        return
+    fn = getattr(ledger, "start_event_wait", None)
+    if not callable(fn):
+        return
+    try:
+        fn(work_order_no=no, wait_kind=kind)
+    except Exception:
+        logger.exception("start_event_wait 失败 wo=%s kind=%s", no, kind)
+
+
+def clear_wait(ledger: Any, *, work_order_no: str) -> None:
+    """清空当前等待，不改事件异常摘要。失败只打日志。"""
+    no = (work_order_no or "").strip()
+    if not no or ledger is None:
+        return
+    fn = getattr(ledger, "clear_event_wait", None)
+    if not callable(fn):
+        return
+    try:
+        fn(work_order_no=no)
+    except Exception:
+        logger.exception("clear_event_wait 失败 wo=%s", no)
+
+
 def pick_mentions(
     *,
     wait_kind: str,

inquiry-agent/tests/test_event_exception.py

@@ -103,6 +103,33 @@ class TestMentions(unittest.TestCase):
         self.assertEqual(conversation_of(wait_kind=WAIT_DEAL, collab_chat_id=""), "private")
 
 
+class TestMemoryEventWait(unittest.TestCase):
+    def test_same_kind_keeps_start(self) -> None:
+        from agent.ledger.memory_ledger import MemoryLedger
+
+        book = MemoryLedger()
+        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")
+        first = book.get_ticket(work_order_no=wo).event_wait_started_at
+        book.start_event_wait(work_order_no=wo, wait_kind="product_quote")
+        self.assertEqual(book.get_ticket(work_order_no=wo).event_wait_started_at, first)
+
+    def test_record_appends_and_alerts(self) -> None:
+        from agent.ledger.memory_ledger import MemoryLedger
+
+        book = MemoryLedger()
+        created = book.create_ticket(sender_id="s1", business_line="AIR", facts={})
+        wo = created["work_order_no"]
+        book.start_event_wait(work_order_no=wo, wait_kind="air_quote")
+        book.record_event_exception(work_order_no=wo, event_exception="航线响应超时")
+        book.record_event_exception(work_order_no=wo, event_exception="销售响应超时")
+        t = book.get_ticket(work_order_no=wo)
+        self.assertEqual(t.event_exception, "航线响应超时；销售响应超时")
+        self.assertEqual(t.event_wait_alerted, 1)
+        self.assertEqual(book.list_event_waits(), [])
+
+
 class TestEventExceptionCopy(unittest.TestCase):
     def test_group_air(self) -> None:
         text = copy.event_exception_session_text(