From bc1b45a8fde4ef1cf61bcd04ea7f1d8667b1270d Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 3 Jun 2026 01:36:45 +0800 Subject: [PATCH] fix(chat,channels): admin session delete, weixin inbound, PG and docs - List/delete chat sessions for administrator by username (cross-tenant UUID) - Return 404 when delete does not remove a session; add tests - Weixin: dispatch lang, inbound attachments, reply persist, native reply timeout - PG compat scrub and administrator session delete repo path - RUNBOOK: WhatsApp re-bind with Remove-Item auth; weixin poll diag scripts Co-authored-by: Cursor --- docs/RUNBOOK.md | 27 +- interfaces/admin/chat_api.py | 73 ++++-- interfaces/admin/routes.py | 52 +++- interfaces/admin/static/app.js | 18 +- interfaces/http/weixin_ilink_api.py | 38 ++- .../application/gateway/inbound_service.py | 232 +++++++++++++++--- runtime/operations/scripts/weixin_install.ps1 | 2 +- .../operations/scripts/weixin_poll_diag.ps1 | 28 +++ runtime/operations/scripts/weixin_start.ps1 | 4 +- .../weixin_bridge/official_runner.ts | 93 ++++++- runtime/operations/weixin_bridge/poll_diag.ts | 109 ++++++++ svc/persistence/pg_compat.py | 25 +- svc/persistence/sa_repos/chat_sessions.py | 134 ++++++++++ svc/persistence/sqlite_store.py | 36 ++- tests/test_admin_auth_rbac.py | 32 ++- tests/test_channel_dispatch_defaults.py | 12 +- ...hat_administrator_cross_tenant_sessions.py | 78 ++++++ tests/test_inbound_service_reply_suppress.py | 111 ++++++++- tests/test_pg_compat.py | 6 + tests/test_weixin_ilink_api.py | 32 +++ 20 files changed, 1056 insertions(+), 86 deletions(-) create mode 100644 runtime/operations/scripts/weixin_poll_diag.ps1 create mode 100644 runtime/operations/weixin_bridge/poll_diag.ts create mode 100644 tests/test_chat_administrator_cross_tenant_sessions.py diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index c27de902..21fd17a8 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -280,8 +280,33 @@ powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\whatsapp_s 说明： - `whatsapp_login.ps1` 会在控制台打印二维码，请用 WhatsApp 手机端的“关联设备”扫码完成绑定。 -- 登录态会落盘在 `data/channel_sidecar/whatsapp/state/auth/`，重启后无需重复扫码。 +- 登录态会落盘在 `data/channel_sidecar/whatsapp/state/auth/`，重启 sidecar 后**一般无需重复扫码**。 - sidecar 收到消息后会调用本地网关 `POST /inbound/whatsapp` 获取 `replies[]` 并回发。 +- sidecar 日志出现 `logged out` 时，需删除上述 `auth` 目录后再执行 `whatsapp_login.ps1`（见下节）。 + +#### 重新绑定设备（换号 / 手机端已解除关联 / 要出新二维码） + +与首次安装不同：**必须先清掉旧登录态**，否则 `whatsapp_login.ps1` 可能不会出现二维码。 + +在仓库根目录执行： + +```powershell +powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\whatsapp_stop.ps1 -Force +Remove-Item -Recurse -Force .\data\channel_sidecar\whatsapp\state\auth +powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\whatsapp_login.ps1 +powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\whatsapp_start.ps1 +powershell -ExecutionPolicy Bypass -File .\runtime\operations\scripts\whatsapp_status.ps1 +``` + +可选：在手机 WhatsApp **设置 → 已关联的设备** 中删除旧的 “oclaw” 设备，再扫码。 + +#### 重新绑定 oclaw 用户（控制台渠道绑定） + +设备已连上、只需把某个 WhatsApp 联系人归属到团队用户时： + +1. 打开 `http://127.0.0.1:8787/admin` → **用户/渠道绑定**，渠道选 `whatsapp`。 +2. 生成绑定码；用该 WhatsApp 向机器人发送：`bind <绑定码>`（与微信相同）。 +3. 无需删除 `state/auth`（那是设备登录态，不是用户归属）。 Admin 可视化调度（新增）： diff --git a/interfaces/admin/chat_api.py b/interfaces/admin/chat_api.py index 65a0436c..dc403b92 100644 --- a/interfaces/admin/chat_api.py +++ b/interfaces/admin/chat_api.py @@ -113,6 +113,7 @@ _DISPATCH_REASON_LABELS_SETTING_KEY = "AIA_DISPATCH_REASON_LABELS_JSON" _SPECIALIST_FLAGS_SETTING_KEY = "AIA_CHAT_SPECIALIST_FLAGS_JSON" _CHANNEL_DISPATCH_INTERACTION_KEY_PREFIX = "channel.dispatch.interaction_mode." _CHANNEL_DISPATCH_SPECIALIST_KEY_PREFIX = "channel.dispatch.specialist." +_CHANNEL_DISPATCH_LANG_KEY_PREFIX = "channel.dispatch.lang." def _channel_dispatch_interaction_key(channel: str) -> str: @@ -122,6 +123,16 @@ def _channel_dispatch_interaction_key(channel: str) -> str: def _channel_dispatch_specialist_key(channel: str) -> str: return f"{_CHANNEL_DISPATCH_SPECIALIST_KEY_PREFIX}{str(channel or '').strip().lower()}" +def _channel_dispatch_lang_key(channel: str) -> str: + return f"{_CHANNEL_DISPATCH_LANG_KEY_PREFIX}{str(channel or '').strip().lower()}" + + +def _normalize_channel_dispatch_lang(raw: Any) -> str: + v = str(raw or "").strip().lower() + if v in {"auto", "zh", "en"}: + return v + return "auto" + def _normalize_channel_dispatch_channel(raw: Any) -> str: ch = str(raw or "").strip().lower() @@ -361,11 +372,7 @@ def _chat_username(ctx: dict[str, Any]) -> str: def _is_administrator_chat_viewer(ctx: dict[str, Any]) -> bool: - """``administrator`` 账户：在单会话读写/导出等接口上可按租户打开任意会话（供审计与 Session Monitor）。 - - ``GET /chat/sessions`` **不再**使用租户全量列表，避免与普通用户会话混在同一侧边栏； - 看全租户会话请用审计页、``GET /admin/api/chat/admin/sessions`` 等专用接口。 - """ + """``administrator`` 账户：可跨租户列出/打开本人名下所有 ``ui_session_owner`` 会话（PG 迁移后历史可能在 smoke/Team 等多租户）。""" return _chat_username(ctx) == "administrator" @@ -377,9 +384,16 @@ def _require_administrator_chat_viewer(ctx: dict[str, Any]) -> None: def _resolve_chat_session(store: SqliteStore, ctx: dict[str, Any], session_id: str): tenant_id = str(ctx.get("tenant_id") or "") user_id = str(ctx.get("user_id") or "") + sid = str(session_id or "").strip() if _is_administrator_chat_viewer(ctx): - return store.get_session_in_tenant(session_id=str(session_id or "").strip(), tenant_id=tenant_id) - return store.get_session_for_user(session_id=session_id, tenant_id=tenant_id, user_id=user_id) + sess = store.get_session_for_administrator_username( + session_id=sid, + username=_chat_username(ctx), + ) + if sess is not None: + return sess + return store.get_session_in_tenant(session_id=sid, tenant_id=tenant_id) + return store.get_session_for_user(session_id=sid, tenant_id=tenant_id, user_id=user_id) def _resolve_chat_session_allow_claim_orphan( @@ -875,12 +889,17 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor ctx = resolve_auth(store, authorization) tenant_id = str(ctx.get("tenant_id") or "") user_id = str(ctx.get("user_id") or "") - # 含 administrator：侧边栏只列「当前登录用户」名下会话，避免租户内他人会话出现在 /chat。 - # 全租户会话列表见审计、GET /admin/api/chat/admin/sessions 等。 - meta = store.get_sessions_list_meta_for_user(tenant_id=tenant_id, user_id=user_id) - rows = store.list_sessions_for_user( - tenant_id=tenant_id, user_id=user_id, limit=limit, offset=offset - ) + if _is_administrator_chat_viewer(ctx): + uname = _chat_username(ctx) + meta = store.get_sessions_list_meta_for_administrator_username(username=uname) + rows = store.list_sessions_for_administrator_username( + username=uname, limit=limit, offset=offset + ) + else: + meta = store.get_sessions_list_meta_for_user(tenant_id=tenant_id, user_id=user_id) + rows = store.list_sessions_for_user( + tenant_id=tenant_id, user_id=user_id, limit=limit, offset=offset + ) return { "ok": True, "total": int(meta.session_count or 0), @@ -962,12 +981,23 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor if not sess: raise HTTPException(status_code=404, detail="session_not_found") if _is_administrator_chat_viewer(ctx): - store.delete_session_in_tenant(session_id=session_id, tenant_id=tenant_id) + deleted = store.delete_session_for_administrator_username( + session_id=session_id, username=_chat_username(ctx) + ) else: - store.delete_session_for_user(session_id=session_id, tenant_id=tenant_id, user_id=user_id) - remaining = store.list_sessions_for_user( - tenant_id=tenant_id, user_id=user_id, limit=1, offset=0 - ) + deleted = store.delete_session_for_user( + session_id=session_id, tenant_id=tenant_id, user_id=user_id + ) + if not deleted: + raise HTTPException(status_code=404, detail="session_not_found") + if _is_administrator_chat_viewer(ctx): + remaining = store.list_sessions_for_administrator_username( + username=_chat_username(ctx), limit=1, offset=0 + ) + else: + remaining = store.list_sessions_for_user( + tenant_id=tenant_id, user_id=user_id, limit=1, offset=0 + ) next_id = str(remaining[0].id) if remaining else "" if not next_id: lang = _api_lang(store) @@ -1563,12 +1593,16 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor specialist = normalize_requested_specialist( store.get_setting(_channel_dispatch_specialist_key(ch)) or "generalist" ) + lang = _normalize_channel_dispatch_lang( + store.get_setting(_channel_dispatch_lang_key(ch)) or "auto" + ) specialist = _apply_specialist_flags(store, specialist) return { "ok": True, "channel": ch, "interaction_mode": interaction_mode, "specialist": specialist, + "lang": lang, "available_specialists": [sid for sid in _chat_specialist_ids() if bool(_specialist_flags_with_overrides(store).get(sid, True))], } @@ -1585,14 +1619,17 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor ch = _normalize_channel_dispatch_channel(channel) interaction_mode = normalize_interaction_mode(body.get("interaction_mode") or "expert") specialist = normalize_requested_specialist(body.get("specialist") or "generalist") + lang = _normalize_channel_dispatch_lang(body.get("lang") or "auto") specialist = _apply_specialist_flags(store, specialist) store.set_setting(_channel_dispatch_interaction_key(ch), interaction_mode) store.set_setting(_channel_dispatch_specialist_key(ch), specialist) + store.set_setting(_channel_dispatch_lang_key(ch), lang) return { "ok": True, "channel": ch, "interaction_mode": interaction_mode, "specialist": specialist, + "lang": lang, } @chat.get("/settings/attachment-limits") diff --git a/interfaces/admin/routes.py b/interfaces/admin/routes.py index b6e2732f..d9bdc65c 100644 --- a/interfaces/admin/routes.py +++ b/interfaces/admin/routes.py @@ -333,6 +333,48 @@ def build_admin_router() -> APIRouter: return parts[1].strip() return "" + def _resolve_login_tenant_id(store: SqliteStore, *, username: str, tenant_id: str) -> str: + """Pick tenant when login body omits tenant_id. + + Chat login sends tenant_id=\"\". Using list_tenants(limit=1) alone picks the newest tenant + (often pg-smoke), which has no chat history. For administrator, prefer the Team tenant, + then the tenant with the most owned sessions. + """ + explicit = str(tenant_id or "").strip() + if explicit: + return explicit + uname = str(username or "").strip().lower() + tenants = store.list_tenants(limit=500) + if not tenants: + return "" + if uname == "administrator": + for row in tenants: + tid = str((row or {}).get("id") or "").strip() + if not tid: + continue + if str((row or {}).get("name") or "").strip().lower() != "team": + continue + if store.get_user_by_username(tenant_id=tid, username="administrator"): + return tid + best_tid = "" + best_n = -1 + for row in tenants: + tid = str((row or {}).get("id") or "").strip() + if not tid: + continue + user = store.get_user_by_username(tenant_id=tid, username="administrator") + if not user: + continue + uid = str(user.get("id") or "") + meta = store.get_sessions_list_meta_for_user(tenant_id=tid, user_id=uid) + n = int(meta.session_count or 0) + if n > best_n: + best_n = n + best_tid = tid + if best_tid: + return best_tid + return str((tenants[0] or {}).get("id") or "").strip() + def _resolve_auth(store: SqliteStore, authorization: str | None) -> dict[str, Any]: token = _extract_bearer(authorization) if not token: @@ -3751,11 +3793,6 @@ def build_admin_router() -> APIRouter: payload = payload or {} store = get_assistant_store() _ensure_admin_bootstrap(store) - tenant_id = str(payload.get("tenant_id") or "").strip() - if not tenant_id: - tenants = store.list_tenants(limit=1) - if tenants: - tenant_id = str((tenants[0] or {}).get("id") or "").strip() purpose = str(payload.get("purpose") or "console").strip().lower() if purpose not in {"console", "chat"}: purpose = "console" @@ -3763,6 +3800,11 @@ def build_admin_router() -> APIRouter: username = str(payload.get("username") or "administrator").strip().lower() else: username = str(payload.get("username") or "").strip().lower() + tenant_id = _resolve_login_tenant_id( + store, + username=username, + tenant_id=str(payload.get("tenant_id") or "").strip(), + ) password = str(payload.get("password") or "").strip() if not tenant_id or not password: return {"ok": False, "error": "tenant_id, username, password are required"} diff --git a/interfaces/admin/static/app.js b/interfaces/admin/static/app.js index d6b641fe..91dc2a14 100644 --- a/interfaces/admin/static/app.js +++ b/interfaces/admin/static/app.js @@ -1528,20 +1528,28 @@ async function renderStack() { const createChannelDispatchCard = (channel, title, initial) => { const curMode = String((initial && initial.interaction_mode) || "expert").trim() || "expert"; const curSpecialist = String((initial && initial.specialist) || "generalist").trim() || "generalist"; + const curLang = String((initial && initial.lang) || "auto").trim().toLowerCase() || "auto"; const specialistSel = el("select", { class: "input" }, availableDispatchSpecialists.map((sid) => el("option", { value: sid, text: sid, selected: sid === curSpecialist ? "selected" : undefined }), )); - const status = el("div", { class: "muted", text: `mode=${curMode} specialist=${curSpecialist}` }); + const langSel = el("select", { class: "input" }, [ + el("option", { value: "auto", text: currentLang === "zh" ? "自动" : "Auto", selected: curLang === "auto" ? "selected" : undefined }), + el("option", { value: "zh", text: currentLang === "zh" ? "中文" : "Chinese", selected: curLang === "zh" ? "selected" : undefined }), + el("option", { value: "en", text: currentLang === "zh" ? "英文" : "English", selected: curLang === "en" ? "selected" : undefined }), + ]); + const status = el("div", { class: "muted", text: `mode=${curMode} specialist=${curSpecialist} lang=${curLang}` }); const saveExpertBtn = el("button", { class: "btn", text: currentLang === "zh" ? "绑定专家" : "Bind specialist", onclick: async () => { const specialist = String(specialistSel.value || "generalist").trim() || "generalist"; + const lang = String(langSel.value || "auto").trim().toLowerCase() || "auto"; const resp = await apiPost(`/admin/api/chat/settings/channel-dispatch/${encodeURIComponent(channel)}`, { interaction_mode: "expert", specialist, + lang, }); - status.textContent = `mode=${String(resp.interaction_mode || "expert")} specialist=${String(resp.specialist || specialist)}`; + status.textContent = `mode=${String(resp.interaction_mode || "expert")} specialist=${String(resp.specialist || specialist)} lang=${String(resp.lang || lang)}`; }, }); const saveComprehensiveBtn = el("button", { @@ -1549,11 +1557,13 @@ async function renderStack() { text: currentLang === "zh" ? "综合" : "Comprehensive", onclick: async () => { const specialist = String(specialistSel.value || "generalist").trim() || "generalist"; + const lang = String(langSel.value || "auto").trim().toLowerCase() || "auto"; const resp = await apiPost(`/admin/api/chat/settings/channel-dispatch/${encodeURIComponent(channel)}`, { interaction_mode: "comprehensive", specialist, + lang, }); - status.textContent = `mode=${String(resp.interaction_mode || "comprehensive")} specialist=${String(resp.specialist || specialist)}`; + status.textContent = `mode=${String(resp.interaction_mode || "comprehensive")} specialist=${String(resp.specialist || specialist)} lang=${String(resp.lang || lang)}`; }, }); return el("div", { class: "card" }, [ @@ -1561,6 +1571,8 @@ async function renderStack() { el("div", { class: "row" }, [ el("label", { text: currentLang === "zh" ? "专家" : "Specialist" }), specialistSel, + el("label", { text: currentLang === "zh" ? "语言" : "Lang" }), + langSel, saveExpertBtn, saveComprehensiveBtn, ]), diff --git a/interfaces/http/weixin_ilink_api.py b/interfaces/http/weixin_ilink_api.py index 5cb094aa..30ae8240 100644 --- a/interfaces/http/weixin_ilink_api.py +++ b/interfaces/http/weixin_ilink_api.py @@ -49,6 +49,40 @@ def _now_ms() -> int: return int(time.time() * 1000) +def _native_reply_timeout_sec() -> float: + raw = str(os.getenv("AIA_CHANNEL_NATIVE_REPLY_TIMEOUT_SEC") or "90").strip() + try: + value = float(raw) + except ValueError: + value = 90.0 + return max(10.0, min(value, 600.0)) + + +async def _run_native_reply_usecase(payload: dict[str, Any]) -> dict[str, Any]: + timeout = _native_reply_timeout_sec() + try: + return await asyncio.wait_for( + asyncio.to_thread(_process_inbound_payload_usecase, payload), + timeout=timeout, + ) + except asyncio.TimeoutError: + channel = _normalize_channel(payload.get("channel")) + chat_id = str(payload.get("chat_id") or payload.get("user_id") or "").strip() + return { + "ok": False, + "error": "native_reply_timeout", + "replies": [ + { + "channel": channel, + "chat_id": chat_id, + "text": "处理超时，请稍后再试。", + "attachments": [], + "metadata": {}, + } + ], + } + + def _normalize_channel(raw: Any) -> str: ch = str(raw or "wechat").strip().lower() return ch if ch else "wechat" @@ -468,7 +502,7 @@ async def weixin_native_reply( return {"ok": False, "error": "missing user_id", "replies": []} if not str(payload.get("account_id") or "").strip(): return {"ok": False, "error": "missing account_id", "replies": []} - out = await asyncio.to_thread(_process_inbound_payload_usecase, payload) + out = await _run_native_reply_usecase(payload) replies = out.get("replies") if isinstance(out, dict) else [] if not isinstance(replies, list): replies = [] @@ -490,7 +524,7 @@ async def whatsapp_native_reply( return {"ok": False, "error": "missing user_id", "replies": []} if not str(payload.get("account_id") or "").strip(): return {"ok": False, "error": "missing account_id", "replies": []} - out = await asyncio.to_thread(_process_inbound_payload_usecase, payload) + out = await _run_native_reply_usecase(payload) replies = out.get("replies") if isinstance(out, dict) else [] if not isinstance(replies, list): replies = [] diff --git a/runtime/application/gateway/inbound_service.py b/runtime/application/gateway/inbound_service.py index 7d6b12a4..8773970c 100644 --- a/runtime/application/gateway/inbound_service.py +++ b/runtime/application/gateway/inbound_service.py @@ -11,6 +11,7 @@ from svc.persistence.assistant_store import get_assistant_store _CHANNEL_DISPATCH_INTERACTION_KEY_PREFIX = "channel.dispatch.interaction_mode." _CHANNEL_DISPATCH_SPECIALIST_KEY_PREFIX = "channel.dispatch.specialist." +_CHANNEL_DISPATCH_LANG_KEY_PREFIX = "channel.dispatch.lang." def _channel_dispatch_interaction_key(channel: str) -> str: @@ -20,28 +21,65 @@ def _channel_dispatch_interaction_key(channel: str) -> str: def _channel_dispatch_specialist_key(channel: str) -> str: return f"{_CHANNEL_DISPATCH_SPECIALIST_KEY_PREFIX}{str(channel or '').strip().lower()}" +def _channel_dispatch_lang_key(channel: str) -> str: + return f"{_CHANNEL_DISPATCH_LANG_KEY_PREFIX}{str(channel or '').strip().lower()}" -def _resolve_channel_dispatch(store: Any, *, channel: str, account: dict[str, Any] | None) -> tuple[str, str]: + +def _normalize_channel_dispatch_lang(raw: Any) -> str: + v = str(raw or "").strip().lower() + if v in {"auto", "zh", "en"}: + return v + return "auto" + + +def _channel_dispatch_setting_aliases(channel: str) -> tuple[str, ...]: ch = str(channel or "").strip().lower() - interaction_mode = normalize_interaction_mode(store.get_setting(_channel_dispatch_interaction_key(ch)) or "expert") - specialist = normalize_requested_specialist(store.get_setting(_channel_dispatch_specialist_key(ch)) or "generalist") + if ch == "wechat": + return ("wechat", "weixin") + if ch == "weixin": + return ("weixin", "wechat") + return (ch,) + + +def _get_channel_dispatch_setting(store: Any, key_prefix: str, channel: str) -> str: + for alias in _channel_dispatch_setting_aliases(channel): + val = str(store.get_setting(f"{key_prefix}{alias}") or "").strip() + if val: + return val + return "" + + +def _resolve_channel_dispatch(store: Any, *, channel: str, account: dict[str, Any] | None) -> tuple[str, str, str]: + ch = str(channel or "").strip().lower() + interaction_mode = normalize_interaction_mode( + _get_channel_dispatch_setting(store, _CHANNEL_DISPATCH_INTERACTION_KEY_PREFIX, ch) or "expert" + ) + specialist = normalize_requested_specialist( + _get_channel_dispatch_setting(store, _CHANNEL_DISPATCH_SPECIALIST_KEY_PREFIX, ch) or "generalist" + ) + lang = _normalize_channel_dispatch_lang( + _get_channel_dispatch_setting(store, _CHANNEL_DISPATCH_LANG_KEY_PREFIX, ch) or "auto" + ) cfg = (account or {}).get("config") if isinstance(cfg, dict): cfg_mode = cfg.get("interaction_mode") cfg_specialist = cfg.get("specialist") + cfg_lang = cfg.get("lang") if cfg_mode is not None: interaction_mode = normalize_interaction_mode(cfg_mode) if cfg_specialist is not None: specialist = normalize_requested_specialist(cfg_specialist) - return interaction_mode, specialist + if cfg_lang is not None: + lang = _normalize_channel_dispatch_lang(cfg_lang) + return interaction_mode, specialist, lang -def _build_admin_gateway_executor(store: Any, *, tenant_id: str, specialist: str, session_id: str) -> Any: +def _build_admin_gateway_executor(store: Any, *, tenant_id: str, specialist: str, session_id: str, lang: str) -> Any: from runtime.agents.factory import build_gateway_executor return build_gateway_executor( store, - lang="zh", + lang=str(lang or "zh"), specialist=specialist, viewer_user_id=None, viewer_username="administrator", @@ -301,6 +339,45 @@ def _parse_generic_inbound(channel_name: str, payload: dict[str, Any]) -> Inboun ) +def _channel_attachments_for_gateway(raw: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Ingest sidecar local_path blobs into attachment store refs for handle_turn.""" + from pathlib import Path + + from svc.files.attachment_assets import AttachmentAssetStore + + out: list[dict[str, Any]] = [] + ast = AttachmentAssetStore() + for a in raw or []: + if not isinstance(a, dict): + continue + aid = str(a.get("attachment_id") or a.get("attachmentId") or "").strip().lower() + if aid: + t = str(a.get("type") or "").strip().lower() or "binary_ref" + out.append({"type": t, "attachment_id": aid}) + continue + lp = str(a.get("local_path") or a.get("media_path") or "").strip() + if not lp: + continue + p = Path(lp) + if not p.is_file(): + continue + kind = str(a.get("kind") or "").strip().lower() + mime = str(a.get("media_type") or a.get("mime") or a.get("mime_type") or "").strip() + if not mime: + mime = "application/octet-stream" + try: + meta = ast.save_bytes(p.read_bytes(), filename=p.name, mime=mime) + except Exception: + continue + if kind == "image" or mime.startswith("image/"): + out.append({"type": "image_ref", "attachment_id": meta.attachment_id}) + elif kind == "video" or mime.startswith("video/"): + out.append({"type": "video_ref", "attachment_id": meta.attachment_id}) + else: + out.append({"type": "binary_ref", "attachment_id": meta.attachment_id}) + return out + + def _parse_message_attachments(raw: Any) -> list[dict[str, Any]]: if raw is None: return [] @@ -330,10 +407,22 @@ def _first_attachment_id(atts: list[dict[str, Any]]) -> str: return "" -def _collect_recent_tool_attachments(*, store: Any, session_id: str) -> list[dict[str, Any]]: - """Fallback for channel delivery: if assistant attachments are missing, reuse recent tool attachments. +def _rows_since_last_user_message(rows: list[Any]) -> list[Any]: + """Return messages after the most recent user turn (current inbound reply scope).""" + last_user_idx = -1 + for i, row in enumerate(rows or []): + role = str(getattr(row, "role", "") or "").strip().lower() + if role == "user": + last_user_idx = i + if last_user_idx < 0: + return list(rows or []) + return list(rows[last_user_idx + 1 :]) - This is intentionally conservative: only returns attachments when they look like media or refs. + +def _collect_recent_tool_attachments(*, store: Any, session_id: str) -> list[dict[str, Any]]: + """Fallback for channel delivery: reuse tool media produced during the current user turn only. + + Avoids re-sending images from earlier conversation turns when the latest assistant row has no attachments. """ sid = str(session_id or "").strip() if not sid: @@ -342,15 +431,14 @@ def _collect_recent_tool_attachments(*, store: Any, session_id: str) -> list[dic rows = store.get_messages(session_id=sid, limit=80) except Exception: rows = [] - # Prefer newest tool attachments. - for row in reversed(list(rows or [])): + scoped = _rows_since_last_user_message(list(rows or [])) + for row in reversed(scoped): role = str(getattr(row, "role", "") or "").strip().lower() if role != "tool": continue atts = _parse_message_attachments(getattr(row, "attachments", None)) if not atts: continue - # Only accept common media/ref shapes. ok = False for a in atts: if not isinstance(a, dict): @@ -447,6 +535,11 @@ def _maybe_expand_reply_attachments_for_channel(reply: dict[str, Any]) -> None: def _collect_reply_attachments_from_history(*, store: Any, session_id: str, reply_text: str) -> list[dict[str, Any]]: + """Attachments for the assistant message whose text equals ``reply_text`` (latest match only). + + Does not fall back to older assistant rows with attachments — that caused WeChat to re-send stale images + on every later text-only reply in the same session. + """ sid = str(session_id or "").strip() if not sid: return [] @@ -455,19 +548,64 @@ def _collect_reply_attachments_from_history(*, store: Any, session_id: str, repl except Exception: rows = [] target = str(reply_text or "").strip() - fallback: list[dict[str, Any]] = [] - matched: list[dict[str, Any]] = [] - for row in rows or []: + if not target: + return [] + for row in reversed(list(rows or [])): role = str(getattr(row, "role", "") or "").strip().lower() if role != "assistant": continue - atts = _parse_message_attachments(getattr(row, "attachments", None)) - if atts: - fallback = atts content = str(getattr(row, "content", "") or "").strip() - if target and content == target and atts: - matched = atts - return matched or fallback + if content != target: + continue + return _parse_message_attachments(getattr(row, "attachments", None)) + return [] + + +def _user_facing_wechat_reply(*, reply: str) -> str: + """Map empty/suppressed provider errors to a short user-visible wechat message.""" + text = str(reply or "").strip() + if not text: + return "暂时无法回复，请稍后再试。" + if _should_suppress_channel_reply(channel="wechat", text=text): + return "模型 API 未配置或不可用，请联系管理员在后台检查大模型配置。" + return text + + +def _latest_user_turn_uuid(store: Any, *, session_id: str) -> str: + rows = store.get_messages(session_id=str(session_id), limit=80) + for m in reversed(rows or []): + if str(getattr(m, "role", "") or "").lower() != "user": + continue + tu = str(getattr(m, "turn_uuid", "") or "").strip() + if tu: + return tu + return "" + + +def _persist_channel_assistant_if_turn_missing( + *, + store: Any, + session_id: str, + turn_uuid: str, + final_text: str, +) -> None: + """Channel inbound may return user-visible text without a persisted assistant row (LLM timeout).""" + from runtime.chat.persist_terminal_fallback import persist_assistant_text_if_turn_missing + + sid = str(session_id or "").strip() + body = str(final_text or "").strip() + if not sid or not body: + return + tu = str(turn_uuid or "").strip() or _latest_user_turn_uuid(store, session_id=sid) + if not tu: + return + persist_assistant_text_if_turn_missing( + store=store, + session_id=sid, + turn_uuid=tu, + final_text=body, + log_prefix="channel_inbound_assistant_persist", + ) def _should_suppress_channel_reply(*, channel: str, text: str) -> bool: @@ -519,6 +657,8 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: text = inbound.text.strip() preface = "" + channel_session_id = "" + channel_turn_uuid = "" if text.lower().startswith("bind "): code = text.split(None, 1)[-1].strip() info = store.consume_bind_code( @@ -591,6 +731,7 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: group_name=group_name, ), ) + channel_session_id = str(session_id) store.ensure_ui_session_owner(session_id=session_id, tenant_id=tenant_id, user_id=user_id) scope = "group" if inbound.is_group else "direct" pe = PolicyEngine() @@ -629,14 +770,25 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: reply = cmd_reply elif not reply: user_text = (inbound.text or "").strip() - if user_text: + gw_attachments = _channel_attachments_for_gateway( + list(inbound.attachments or []) + ) + if not user_text and gw_attachments: + user_text = "用户发送了附件，请根据附件内容回复。" + if user_text or gw_attachments: try: from runtime.gateway import OclawGateway from runtime.types import StandardMessage + from runtime.lang import resolve_runtime_lang - interaction_mode, selected_specialist = _resolve_channel_dispatch( + interaction_mode, selected_specialist, dispatch_lang = _resolve_channel_dispatch( store, channel=inbound.channel, account=account ) + lang = ( + dispatch_lang + if dispatch_lang in {"zh", "en"} + else resolve_runtime_lang(store=store, user_text=user_text) + ) gw = OclawGateway(store=store) msg = StandardMessage( session_id=str(session_id), @@ -645,7 +797,7 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: role=str(role or "member"), channel=str(inbound.channel or "inbound"), text=str(user_text or ""), - attachments=[], + attachments=gw_attachments, metadata={ "tenant_id": tenant_id, "user_id": user_id, @@ -661,22 +813,23 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: tenant_id=tenant_id, specialist="generalist", session_id=str(session_id), + lang=lang, ) specialist_factory = lambda sid: _build_admin_gateway_executor( store, tenant_id=tenant_id, specialist=sid, session_id=str(session_id), + lang=lang, ) - reply = str( - gw.handle_turn( - msg=msg, - lang="zh", - executor=manager, - specialist_executor_factory=specialist_factory, - ).reply_text - or "" - ).strip() + turn_result = gw.handle_turn( + msg=msg, + lang=lang, + executor=manager, + specialist_executor_factory=specialist_factory, + ) + channel_turn_uuid = str(turn_result.turn_uuid or "").strip() + reply = str(turn_result.reply_text or "").strip() reply_attachments = _collect_reply_attachments_from_history( store=store, session_id=str(session_id), @@ -694,9 +847,20 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: else: reply = f"{preface}\n\n{_menu_text()}" - if _should_suppress_channel_reply(channel=inbound.channel, text=reply): + ch_lower = str(inbound.channel or "").strip().lower() + if ch_lower in {"wechat", "weixin"}: + reply = _user_facing_wechat_reply(reply=reply) + elif _should_suppress_channel_reply(channel=inbound.channel, text=reply): return {"ok": True, "replies": []} + if channel_session_id and reply: + _persist_channel_assistant_if_turn_missing( + store=store, + session_id=channel_session_id, + turn_uuid=channel_turn_uuid, + final_text=reply, + ) + if adapter is not None: replies = [adapter.format_outbound(OutboundMessage(external_chat_id=inbound.external_chat_id, text=reply))] else: diff --git a/runtime/operations/scripts/weixin_install.ps1 b/runtime/operations/scripts/weixin_install.ps1 index 64ed1d26..035403b5 100644 --- a/runtime/operations/scripts/weixin_install.ps1 +++ b/runtime/operations/scripts/weixin_install.ps1 @@ -15,7 +15,7 @@ $sidecarRoot = Join-Path $oclawRoot "data\\channel_sidecar\\$ChannelId" $stateDir = Join-Path $sidecarRoot "state" function Sync-WeixinBridgeRunners { $bridgeSrc = Join-Path $oclawRoot "runtime\\operations\\weixin_bridge" - foreach ($name in @("official_runner.ts", "login.ts")) { + foreach ($name in @("official_runner.ts", "login.ts", "poll_diag.ts")) { $srcPath = Join-Path $bridgeSrc $name if (-not (Test-Path $srcPath)) { throw "missing bridge source file: $srcPath" diff --git a/runtime/operations/scripts/weixin_poll_diag.ps1 b/runtime/operations/scripts/weixin_poll_diag.ps1 new file mode 100644 index 00000000..f29c9f79 --- /dev/null +++ b/runtime/operations/scripts/weixin_poll_diag.ps1 @@ -0,0 +1,28 @@ +param( + [switch]$ResetCursor = $false +) + +$ErrorActionPreference = "Stop" +$repoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..\..\..")).Path +$sidecarRoot = Join-Path $repoRoot "data\channel_sidecar\oclaw-weixin" +$stateDir = Join-Path $sidecarRoot "state" +$bridgeSrc = Join-Path $repoRoot "runtime\operations\weixin_bridge\poll_diag.ts" + +if (-not (Test-Path $bridgeSrc)) { + throw "missing $bridgeSrc" +} +Copy-Item -Path $bridgeSrc -Destination (Join-Path $sidecarRoot "poll_diag.ts") -Force + +$args = @("exec", "--", "tsx", "poll_diag.ts") +if ($ResetCursor) { + $args += "--reset-cursor" +} + +Push-Location $sidecarRoot +try { + $env:OCLAW_STATE_DIR = $stateDir + $env:OPENCLAW_STATE_DIR = $stateDir + npm.cmd @args +} finally { + Pop-Location +} diff --git a/runtime/operations/scripts/weixin_start.ps1 b/runtime/operations/scripts/weixin_start.ps1 index 72f354d2..af914610 100644 --- a/runtime/operations/scripts/weixin_start.ps1 +++ b/runtime/operations/scripts/weixin_start.ps1 @@ -84,7 +84,7 @@ $cleaned = Stop-SidecarProcesses Remove-Item -Force $pidFile -ErrorAction SilentlyContinue if (Test-Path $bridgeSrc) { - foreach ($name in @("official_runner.ts", "login.ts")) { + foreach ($name in @("official_runner.ts", "login.ts", "poll_diag.ts")) { $srcPath = Join-Path $bridgeSrc $name if (Test-Path $srcPath) { Copy-Item -Path $srcPath -Destination (Join-Path $sidecarRoot $name) -Force @@ -99,7 +99,7 @@ if (Test-Path (Join-Path $sidecarRoot "official_runner.ts")) { $cmd = "cmd.exe" $args = @( "/c", - "cd /d $sidecarRoot && set OCLAW_STATE_DIR=$stateDir&& set AIA_GATEWAY_BASE_URL=$GatewayBaseUrl&& set NODE_PATH=$sidecarRoot\node_modules&& npm.cmd exec -- tsx official_runner.ts" + "cd /d $sidecarRoot && set OCLAW_STATE_DIR=$stateDir&& set OPENCLAW_STATE_DIR=$stateDir&& set AIA_GATEWAY_BASE_URL=$GatewayBaseUrl&& set OCLAW_WEIXIN_LOG_FILE=$logPath&& set NODE_PATH=$sidecarRoot\node_modules&& npm.cmd exec -- tsx official_runner.ts" ) $p = Start-Process -FilePath $cmd -ArgumentList $args -WorkingDirectory $sidecarRoot -PassThru -WindowStyle Hidden -RedirectStandardOutput $logPath -RedirectStandardError $errPath Set-Content -Path $pidFile -Value $p.Id diff --git a/runtime/operations/weixin_bridge/official_runner.ts b/runtime/operations/weixin_bridge/official_runner.ts index 5736e7c0..aadd95a8 100644 --- a/runtime/operations/weixin_bridge/official_runner.ts +++ b/runtime/operations/weixin_bridge/official_runner.ts @@ -8,6 +8,7 @@ type TokenMap = Record; const LOCAL_BASE_URL = (process.env.AIA_GATEWAY_BASE_URL || "http://127.0.0.1:8787").trim(); const STATE_DIR = (process.env.OCLAW_STATE_DIR || path.resolve(process.cwd(), "state")).trim(); const STATE_FILE = path.join(STATE_DIR, "official_bridge_state.json"); +const LOG_FILE = String(process.env.OCLAW_WEIXIN_LOG_FILE || "").trim(); const POLL_TIMEOUT_MS = 35_000; const DEFAULT_CDN_BASE_URL = "https://novac2c.cdn.weixin.qq.com/c2c"; @@ -65,7 +66,21 @@ class HttpStatusError extends Error { } function log(msg: string): void { - process.stdout.write(`${new Date().toISOString()} [official-weixin] ${msg}\n`); + const line = `${new Date().toISOString()} [official-weixin] ${msg}\n`; + // When started via Start-Process -RedirectStandardOutput, Node may block-buffer stdout; + // append directly so operators see poll/inbound lines in weixin_sidecar.log immediately. + if (LOG_FILE) { + try { + fs.appendFileSync(LOG_FILE, line, "utf8"); + } catch { + // ignore + } + } + try { + process.stdout.write(line); + } catch { + // ignore + } } function ensureDir(dir: string): void { @@ -88,6 +103,18 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +async function withTimeout(promise: Promise, ms: number, label: string): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${label} timeout after ${ms}ms`)), ms); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + if (timer) clearTimeout(timer); + } +} + function resolvePluginRoot(): string { const configured = String(process.env.OCLAW_WEIXIN_PLUGIN_ROOT || "").trim(); return configured || path.join(process.cwd(), "node_modules", "@tencent-weixin", "openclaw-weixin"); @@ -167,10 +194,12 @@ function resolveHeaders(token: string): Record { async function postNativeReply(token: string, body: Json): Promise { const url = `${LOCAL_BASE_URL.replace(/\/+$/, "")}/weixin/native/reply`; + const timeoutMs = Number(process.env.OCLAW_WEIXIN_NATIVE_REPLY_TIMEOUT_MS || "100000") || 100000; const res = await fetch(url, { method: "POST", headers: resolveHeaders(token), body: JSON.stringify(body), + signal: AbortSignal.timeout(timeoutMs), }); const text = await res.text(); if (!res.ok) { @@ -330,9 +359,19 @@ async function handleInboundMessage( ): Promise { const fromUser = String(params.full.from_user_id || "").trim(); const toUser = String(params.full.to_user_id || "").trim(); - if (!fromUser) return params.localCursor; - if (toUser && toUser === fromUser) return params.localCursor; - if (Number(params.full.message_type || 1) !== 1) return params.localCursor; + if (!fromUser) { + log(`skip inbound: missing from_user_id`); + return params.localCursor; + } + if (toUser && toUser === fromUser) { + log(`skip inbound: self-message from=${fromUser}`); + return params.localCursor; + } + if (Number(params.full.message_type || 1) !== 1) { + log(`skip inbound: message_type=${String(params.full.message_type ?? "")} from=${fromUser}`); + return params.localCursor; + } + log(`inbound from=${fromUser} to=${toUser || "-"}`); const contextToken = String(params.full.context_token || "").trim(); if (contextToken) { @@ -341,17 +380,33 @@ async function handleInboundMessage( } const mediaItem = pickDownloadableMedia(params.full); - const mediaOpts = mediaItem - ? await modules.downloadMediaFromItem(mediaItem, { - cdnBaseUrl: DEFAULT_CDN_BASE_URL, - saveMedia: saveMediaBuffer, - log: (msg: string) => log(`media ${msg}`), - errLog: (msg: string) => log(`media-error ${msg}`), - label: "inbound", - }) - : {}; + let mediaOpts: Json = {}; + if (mediaItem) { + log("media download start"); + try { + mediaOpts = await withTimeout( + modules.downloadMediaFromItem(mediaItem, { + cdnBaseUrl: DEFAULT_CDN_BASE_URL, + saveMedia: saveMediaBuffer, + log: (msg: string) => log(`media ${msg}`), + errLog: (msg: string) => log(`media-error ${msg}`), + label: "inbound", + }), + Number(process.env.OCLAW_WEIXIN_MEDIA_TIMEOUT_MS || "90000") || 90000, + "media download", + ); + log("media download done"); + } catch (err) { + log(`media download failed err=${String(err)}`); + mediaOpts = {}; + } + } const ctx = modules.weixinMessageToMsgContext(params.full, params.accountId, mediaOpts); + const bodyText = String((ctx as Json).Body || "").trim(); + const attachCount = buildAttachmentsFromMedia(mediaOpts).length; + log(`native call bodyLen=${bodyText.length} attachments=${attachCount}`); let replies: Json[] = []; + const tNative0 = Date.now(); try { const native = await postNativeReply(params.token, { channel: "wechat", @@ -366,6 +421,11 @@ async function handleInboundMessage( }, }); replies = Array.isArray(native.replies) ? (native.replies as Json[]) : []; + const firstText = replies.length ? String((replies[0] as Json).text || "").trim() : ""; + log(`native done ms=${Date.now() - tNative0} replies=${replies.length} firstTextLen=${firstText.length}`); + if (!replies.length) { + log("native returned 0 replies (empty inbound, gateway suppress, or LLM produced no text)"); + } } catch (err) { log(`native reply failed; no fallback enabled err=${String(err)}`); await safeSendNativeFailureNotice(modules, { @@ -447,6 +507,13 @@ async function main(): Promise { timeoutMs: POLL_TIMEOUT_MS + 5000, }); const msgs = Array.isArray(out.msgs) ? (out.msgs as Json[]) : []; + const ret = Number(out.ret ?? 0); + const errcode = out.errcode; + if (ret !== 0 || errcode != null) { + log(`poll ret=${ret} errcode=${String(errcode ?? "")} errmsg=${String(out.errmsg ?? "")}`); + } else if (msgs.length > 0) { + log(`poll batch=${msgs.length}`); + } const nextCloudCursor = String(out.get_updates_buf || cloudCursor || "").trim(); if (nextCloudCursor) { cloudCursor = nextCloudCursor; diff --git a/runtime/operations/weixin_bridge/poll_diag.ts b/runtime/operations/weixin_bridge/poll_diag.ts new file mode 100644 index 00000000..89329ba0 --- /dev/null +++ b/runtime/operations/weixin_bridge/poll_diag.ts @@ -0,0 +1,109 @@ +/** + * One-shot ilink getUpdates probe (diagnose "sidecar started but no messages"). + * Run from repo: runtime/operations/scripts/weixin_poll_diag.ps1 + */ +import fs from "node:fs"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; + +const STATE_DIR = (process.env.OCLAW_STATE_DIR || path.resolve(process.cwd(), "state")).trim(); +const STATE_FILE = path.join(STATE_DIR, "official_bridge_state.json"); + +function resolvePluginRoot(): string { + const configured = String(process.env.OCLAW_WEIXIN_PLUGIN_ROOT || "").trim(); + return configured || path.join(process.cwd(), "node_modules", "@tencent-weixin", "openclaw-weixin"); +} + +async function resolveAccount(): Promise<{ accountId: string; token: string; cloudBaseUrl: string }> { + if (!String(process.env.OPENCLAW_STATE_DIR || "").trim()) { + process.env.OPENCLAW_STATE_DIR = STATE_DIR; + } + const pluginRoot = resolvePluginRoot(); + const srcRoot = path.join(pluginRoot, "src"); + const importTs = async (relativePath: string) => import(pathToFileURL(path.join(srcRoot, relativePath)).href); + const accountsMod = await importTs(path.join("auth", "accounts.ts")); + const listIds = accountsMod.listIndexedWeixinAccountIds as () => string[]; + const loadAcc = accountsMod.loadWeixinAccount as (id: string) => Record; + const accountId = String((listIds() || [])[0] || "").trim(); + if (!accountId) throw new Error("no weixin account; run weixin_login.ps1"); + const data = loadAcc(accountId) || {}; + const token = String(data.token || "").trim(); + if (!token) throw new Error(`missing token for ${accountId}; run weixin_login.ps1`); + const envCloud = String(process.env.OCLAW_WEIXIN_CLOUD_BASE_URL || "").trim(); + const cfgCloud = String(data.baseUrl || "").trim(); + const cloudBaseUrl = (envCloud || cfgCloud || "https://ilinkai.weixin.qq.com").trim(); + return { accountId, token, cloudBaseUrl }; +} + +async function main(): Promise { + const reset = process.argv.includes("--reset-cursor"); + const stateRaw = (() => { + try { + return JSON.parse(fs.readFileSync(STATE_FILE, "utf8")) as Record; + } catch { + return {}; + } + })(); + let cursor = reset ? "" : String(stateRaw.cloud_cursor || "").trim(); + const { accountId, token, cloudBaseUrl } = await resolveAccount(); + const pluginRoot = resolvePluginRoot(); + const apiMod = await import(pathToFileURL(path.join(pluginRoot, "src", "api", "api.ts")).href); + const getUpdates = apiMod.getUpdates as (p: Record) => Promise>; + + console.log( + JSON.stringify( + { + accountId, + cloudBaseUrl, + cursorPrefix: cursor.slice(0, 48), + resetCursor: reset, + }, + null, + 2, + ), + ); + + const out = await getUpdates({ + baseUrl: cloudBaseUrl, + token, + get_updates_buf: cursor, + timeoutMs: 15_000, + }); + const msgs = Array.isArray(out.msgs) ? out.msgs : []; + console.log( + JSON.stringify( + { + ret: out.ret, + errcode: out.errcode, + errmsg: out.errmsg, + msgCount: msgs.length, + nextCursorPrefix: String(out.get_updates_buf || "").slice(0, 48), + }, + null, + 2, + ), + ); + if (msgs.length > 0) { + const m = msgs[0] as Record; + console.log( + "first_msg", + JSON.stringify( + { + from_user_id: m.from_user_id, + to_user_id: m.to_user_id, + message_type: m.message_type, + has_text: Boolean(m.text || (m as { content?: unknown }).content), + }, + null, + 2, + ), + ); + } else { + console.log("hint=send a NEW text message to the bot now, then re-run this script within 30s"); + } +} + +void main().catch((err) => { + console.error("poll_diag_failed", err); + process.exit(1); +}); diff --git a/svc/persistence/pg_compat.py b/svc/persistence/pg_compat.py index a3bbe8d5..011faa23 100644 --- a/svc/persistence/pg_compat.py +++ b/svc/persistence/pg_compat.py @@ -37,6 +37,27 @@ def scrub_nul_bytes_from_jsonable(obj: Any) -> Any: return obj +def escape_percent_in_sql_literals(sql: str) -> str: + """Double ``%`` inside single-quoted literals (psycopg treats ``%`` as placeholder syntax).""" + out: list[str] = [] + i = 0 + in_single = False + while i < len(sql): + ch = sql[i] + if ch == "'" and (i == 0 or sql[i - 1] != "\\"): + in_single = not in_single + out.append(ch) + i += 1 + continue + if ch == "%" and in_single: + out.append("%%") + i += 1 + continue + out.append(ch) + i += 1 + return "".join(out) + + def qmarks_to_percent(sql: str) -> str: """Replace ``?`` placeholders outside single-quoted strings with ``%s`` (psycopg).""" out: list[str] = [] @@ -218,11 +239,13 @@ def rewrite_sqlite_extensions_for_postgres(sql: str) -> str: def adapt_sql_for_postgres(sql: str) -> str: - return qmarks_to_percent(rewrite_sqlite_extensions_for_postgres(sql)) + rewritten = rewrite_sqlite_extensions_for_postgres(sql) + return escape_percent_in_sql_literals(qmarks_to_percent(rewritten)) __all__ = [ "adapt_sql_for_postgres", + "escape_percent_in_sql_literals", "qmarks_to_percent", "rewrite_sqlite_extensions_for_postgres", "scrub_nul_bytes_from_jsonable", diff --git a/svc/persistence/sa_repos/chat_sessions.py b/svc/persistence/sa_repos/chat_sessions.py index 8a4dd921..dc9c2595 100644 --- a/svc/persistence/sa_repos/chat_sessions.py +++ b/svc/persistence/sa_repos/chat_sessions.py @@ -193,6 +193,110 @@ class ChatSessionsSaRepository: else None, ) + def _administrator_username_predicate(self, username: str) -> Any: + uname = str(username or "").strip().lower() + return func.lower(app_user.c.username) == uname + + def list_chat_sessions_for_administrator_username( + self, + *, + username: str, + limit: int | None, + offset: int, + ) -> list[ChatSession]: + """All sessions owned by any ``app_user`` row with this login name (cross-tenant).""" + pred = self._administrator_username_predicate(username) + stmt = ( + select( + chat_session.c.id, + chat_session.c.title, + chat_session.c.created_at, + chat_session.c.last_message_at, + ) + .select_from( + chat_session.join( + ui_session_owner, + ui_session_owner.c.session_id == chat_session.c.id, + ).join( + app_user, + and_( + app_user.c.id == ui_session_owner.c.user_id, + app_user.c.tenant_id == ui_session_owner.c.tenant_id, + ), + ) + ) + .where(pred) + .order_by(*_activity_order()) + ) + if limit is not None: + stmt = stmt.limit(int(limit)).offset(int(offset)) + with self._engine.connect() as conn: + rows = conn.execute(stmt).mappings().all() + return [_session_from_row(r) for r in rows] + + def sessions_list_meta_for_administrator_username(self, *, username: str) -> SessionsListMeta: + pred = self._administrator_username_predicate(username) + with self._engine.connect() as conn: + row = conn.execute( + select( + func.count(distinct(chat_session.c.id)).label("c"), + func.max( + func.coalesce(chat_session.c.last_message_at, chat_session.c.created_at) + ).label("latest_activity_at"), + ) + .select_from( + chat_session.join( + ui_session_owner, + ui_session_owner.c.session_id == chat_session.c.id, + ).join( + app_user, + and_( + app_user.c.id == ui_session_owner.c.user_id, + app_user.c.tenant_id == ui_session_owner.c.tenant_id, + ), + ) + ) + .where(pred) + ).mappings().first() + return SessionsListMeta( + session_count=int(row["c"] or 0) if row else 0, + latest_activity_at=str(row["latest_activity_at"]) + if row and row.get("latest_activity_at") is not None + else None, + ) + + def fetch_chat_session_for_administrator_username( + self, *, session_id: str, username: str + ) -> ChatSession | None: + sid = str(session_id or "").strip() + if not sid: + return None + pred = self._administrator_username_predicate(username) + with self._engine.connect() as conn: + row = conn.execute( + select( + chat_session.c.id, + chat_session.c.title, + chat_session.c.created_at, + chat_session.c.last_message_at, + ) + .select_from( + chat_session.join( + ui_session_owner, + ui_session_owner.c.session_id == chat_session.c.id, + ).join( + app_user, + and_( + app_user.c.id == ui_session_owner.c.user_id, + app_user.c.tenant_id == ui_session_owner.c.tenant_id, + ), + ) + ) + .where(chat_session.c.id == sid, pred) + .limit(1) + ).mappings().first() + return _session_from_row(row) if row else None + def sessions_list_meta_for_user(self, *, tenant_id: str, user_id: str) -> SessionsListMeta: tid, uid = str(tenant_id), str(user_id) with self._engine.connect() as conn: @@ -378,6 +482,36 @@ class ChatSessionsSaRepository: with self._engine.begin() as conn: conn.execute(delete(chat_session).where(chat_session.c.id == sid)) + def try_delete_chat_session_for_administrator_username( + self, *, session_id: str, username: str + ) -> bool: + sid = str(session_id or "").strip() + if not sid: + return False + pred = self._administrator_username_predicate(username) + with self._engine.begin() as conn: + chk = conn.execute( + select(1) + .select_from( + chat_session.join( + ui_session_owner, + ui_session_owner.c.session_id == chat_session.c.id, + ).join( + app_user, + and_( + app_user.c.id == ui_session_owner.c.user_id, + app_user.c.tenant_id == ui_session_owner.c.tenant_id, + ), + ) + ) + .where(chat_session.c.id == sid, pred) + .limit(1) + ).first() + if not chk: + return False + conn.execute(delete(chat_session).where(chat_session.c.id == sid)) + return True + def try_delete_chat_session_for_tenant(self, *, session_id: str, tenant_id: str) -> bool: sid, tid = str(session_id or "").strip(), str(tenant_id) if not sid: diff --git a/svc/persistence/sqlite_store.py b/svc/persistence/sqlite_store.py index 4f742d7d..d86ef595 100644 --- a/svc/persistence/sqlite_store.py +++ b/svc/persistence/sqlite_store.py @@ -1384,6 +1384,32 @@ class SqliteStore: tenant_id=tenant_id, user_id=user_id ) + def list_sessions_for_administrator_username( + self, + *, + username: str, + limit: int | None = None, + offset: int = 0, + ) -> list[ChatSession]: + return self._chat_sessions_repo().list_chat_sessions_for_administrator_username( + username=username, + limit=limit, + offset=int(offset), + ) + + def get_sessions_list_meta_for_administrator_username(self, *, username: str) -> SessionsListMeta: + return self._chat_sessions_repo().sessions_list_meta_for_administrator_username( + username=username + ) + + def get_session_for_administrator_username( + self, *, session_id: str, username: str + ) -> Optional[ChatSession]: + return self._chat_sessions_repo().fetch_chat_session_for_administrator_username( + session_id=session_id, + username=username, + ) + def list_sessions_for_tenant( self, *, @@ -1813,11 +1839,19 @@ class SqliteStore: return int(total_row_c or 0), users, totals def delete_session_in_tenant(self, *, session_id: str, tenant_id: str) -> bool: - """Delete session if it belongs to tenant (used by administrator account).""" + """Delete session if it belongs to tenant.""" return self._chat_sessions_repo().try_delete_chat_session_for_tenant( session_id=session_id, tenant_id=tenant_id ) + def delete_session_for_administrator_username( + self, *, session_id: str, username: str + ) -> bool: + """Delete session owned by any ``app_user`` row with this login name (cross-tenant).""" + return self._chat_sessions_repo().try_delete_chat_session_for_administrator_username( + session_id=session_id, username=username + ) + def delete_session(self, session_id: str) -> None: self._chat_sessions_repo().delete_chat_session_by_id(session_id=session_id) diff --git a/tests/test_admin_auth_rbac.py b/tests/test_admin_auth_rbac.py index d1204547..9da1af0a 100644 --- a/tests/test_admin_auth_rbac.py +++ b/tests/test_admin_auth_rbac.py @@ -10,7 +10,7 @@ from fastapi.testclient import TestClient from interfaces.http.fastapi_app import create_app from svc.persistence.db.engine import clear_assistant_engine_cache from svc.persistence.sqlite_store import SqliteStore -from svc.persistence.assistant_store import get_assistant_store +from svc.persistence.assistant_store import get_assistant_store, reset_assistant_store_singleton class AdminAuthRBACTests(unittest.TestCase): @@ -21,6 +21,10 @@ class AdminAuthRBACTests(unittest.TestCase): import os os.environ["OPS_ASSISTANT_DB_PATH"] = str(db) + os.environ["AIA_ASSISTANT_DB_PATH"] = str(db) + os.environ["AIA_ASSISTANT_DB_BACKEND"] = "sqlite" + os.environ.pop("AIA_ASSISTANT_DATABASE_URL", None) + reset_assistant_store_singleton() os.environ["OPS_ASSISTANT_PASSWORD"] = "test-admin-pass" store = SqliteStore(str(db)) t = store.create_tenant("Team") @@ -38,6 +42,7 @@ class AdminAuthRBACTests(unittest.TestCase): def tearDown(self) -> None: clear_assistant_engine_cache() + reset_assistant_store_singleton() self._tmp.cleanup() def _login(self, username: str = "administrator", password: str = "test-admin-pass") -> str: @@ -123,6 +128,31 @@ class AdminAuthRBACTests(unittest.TestCase): self.assertTrue(data.get("ok"), data) self.assertEqual(str((data.get("session") or {}).get("username") or ""), "carol") + def test_administrator_login_without_tenant_id_prefers_team(self) -> None: + store = get_assistant_store() + newer = store.create_tenant("pg-smoke-newer") + store.create_user_account( + tenant_id=str(newer["id"]), + username="administrator", + display_name="Admin", + role="admin", + password_hash=hashlib.sha256("test-admin-pass".encode("utf-8")).hexdigest(), + is_active=True, + ) + self.client.post("/admin/api/auth/bootstrap", json={}) + resp = self.client.post( + "/admin/api/auth/login", + json={ + "tenant_id": "", + "username": "administrator", + "password": "test-admin-pass", + "purpose": "chat", + }, + ) + data = resp.json() + self.assertTrue(data.get("ok"), data) + self.assertEqual(str((data.get("session") or {}).get("tenant_id") or ""), self.tenant_id) + def test_console_login_allows_member_username_with_admin_read(self) -> None: store = get_assistant_store() store.create_user_account( diff --git a/tests/test_channel_dispatch_defaults.py b/tests/test_channel_dispatch_defaults.py index 184b9420..72c9c344 100644 --- a/tests/test_channel_dispatch_defaults.py +++ b/tests/test_channel_dispatch_defaults.py @@ -15,9 +15,10 @@ class _DummyStore: def test_channel_dispatch_defaults_to_expert_and_generalist() -> None: store = _DummyStore() - interaction_mode, specialist = _resolve_channel_dispatch(store, channel="weixin", account=None) + interaction_mode, specialist, lang = _resolve_channel_dispatch(store, channel="weixin", account=None) assert interaction_mode == "expert" assert specialist == "generalist" + assert lang == "auto" def test_channel_dispatch_uses_global_settings() -> None: @@ -25,11 +26,13 @@ def test_channel_dispatch_uses_global_settings() -> None: { "channel.dispatch.interaction_mode.whatsapp": "comprehensive", "channel.dispatch.specialist.whatsapp": "ops", + "channel.dispatch.lang.whatsapp": "en", } ) - interaction_mode, specialist = _resolve_channel_dispatch(store, channel="whatsapp", account=None) + interaction_mode, specialist, lang = _resolve_channel_dispatch(store, channel="whatsapp", account=None) assert interaction_mode == "comprehensive" assert specialist == "ops" + assert lang == "en" def test_channel_dispatch_account_config_overrides_global() -> None: @@ -37,15 +40,18 @@ def test_channel_dispatch_account_config_overrides_global() -> None: { "channel.dispatch.interaction_mode.weixin": "comprehensive", "channel.dispatch.specialist.weixin": "ops", + "channel.dispatch.lang.weixin": "zh", } ) account = { "config": { "interaction_mode": "expert", "specialist": "generalist", + "lang": "en", } } - interaction_mode, specialist = _resolve_channel_dispatch(store, channel="weixin", account=account) + interaction_mode, specialist, lang = _resolve_channel_dispatch(store, channel="weixin", account=account) assert interaction_mode == "expert" assert specialist == "generalist" + assert lang == "en" diff --git a/tests/test_chat_administrator_cross_tenant_sessions.py b/tests/test_chat_administrator_cross_tenant_sessions.py new file mode 100644 index 00000000..7a514fde --- /dev/null +++ b/tests/test_chat_administrator_cross_tenant_sessions.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import hashlib + +import pytest + +from svc.persistence.db.engine import clear_assistant_engine_cache +from svc.persistence.sqlite_store import SqliteStore +from svc.persistence.assistant_store import reset_assistant_store_singleton + + +@pytest.fixture +def store_two_tenants(monkeypatch: pytest.MonkeyPatch, tmp_path) -> SqliteStore: + monkeypatch.delenv("AIA_ASSISTANT_DATABASE_URL", raising=False) + monkeypatch.setenv("AIA_ASSISTANT_DB_BACKEND", "sqlite") + dbfile = tmp_path / "admin_xtenant.sqlite" + monkeypatch.setenv("AIA_ASSISTANT_DB_PATH", str(dbfile)) + clear_assistant_engine_cache() + reset_assistant_store_singleton() + s = SqliteStore(str(dbfile)) + pwd_hash = hashlib.sha256(b"pass").hexdigest() + t1 = s.create_tenant("Team") + t2 = s.create_tenant("Other") + u1 = s.create_user_account( + tenant_id=str(t1["id"]), + username="administrator", + display_name="Admin", + role="admin", + password_hash=pwd_hash, + is_active=True, + ) + s.create_user_account( + tenant_id=str(t2["id"]), + username="administrator", + display_name="Admin", + role="admin", + password_hash=pwd_hash, + is_active=True, + ) + s.create_session_for_user(title="team-sess", tenant_id=str(t1["id"]), user_id=str(u1["id"])) + s.create_session_for_user( + title="other-sess", + tenant_id=str(t2["id"]), + user_id=str(s.get_user_by_username(tenant_id=str(t2["id"]), username="administrator")["id"]), + ) + yield s + clear_assistant_engine_cache() + reset_assistant_store_singleton() + + +def test_list_sessions_for_administrator_username_cross_tenant( + store_two_tenants: SqliteStore, +) -> None: + s = store_two_tenants + rows = s.list_sessions_for_administrator_username(username="administrator", limit=50) + titles = {x.title for x in rows} + assert "team-sess" in titles + assert "other-sess" in titles + meta = s.get_sessions_list_meta_for_administrator_username(username="administrator") + assert int(meta.session_count or 0) == 2 + + +def test_delete_session_for_administrator_username_cross_tenant( + store_two_tenants: SqliteStore, +) -> None: + s = store_two_tenants + rows = s.list_sessions_for_administrator_username(username="administrator", limit=50) + other = next(x for x in rows if x.title == "other-sess") + team = next(x for x in rows if x.title == "team-sess") + team_owner = s.get_ui_session_owner(session_id=str(team.id)) + login_tid = str((team_owner or {}).get("tenant_id") or "") + assert login_tid + assert s.delete_session_in_tenant(session_id=str(other.id), tenant_id=login_tid) is False + assert s.get_session_for_administrator_username(session_id=str(other.id), username="administrator") is not None + assert s.delete_session_for_administrator_username(session_id=str(other.id), username="administrator") is True + assert s.get_session_for_administrator_username(session_id=str(other.id), username="administrator") is None + meta = s.get_sessions_list_meta_for_administrator_username(username="administrator") + assert int(meta.session_count or 0) == 1 diff --git a/tests/test_inbound_service_reply_suppress.py b/tests/test_inbound_service_reply_suppress.py index 882e4cf6..1bf0be28 100644 --- a/tests/test_inbound_service_reply_suppress.py +++ b/tests/test_inbound_service_reply_suppress.py @@ -5,10 +5,15 @@ from dataclasses import dataclass from runtime.application.gateway.inbound_service import ( _collect_reply_attachments_from_history, _collect_recent_tool_attachments, + _get_channel_dispatch_setting, + _latest_user_turn_uuid, _maybe_add_media_path_for_wechat_reply, _maybe_expand_reply_attachments_for_channel, _parse_message_attachments, + _persist_channel_assistant_if_turn_missing, + _resolve_channel_dispatch, _should_suppress_channel_reply, + _user_facing_wechat_reply, ) @@ -23,6 +28,88 @@ def test_should_not_suppress_non_weixin_channel() -> None: assert _should_suppress_channel_reply(channel="admin_chat", text=text) is False +def test_user_facing_wechat_reply_maps_empty_and_api_key_errors() -> None: + assert _user_facing_wechat_reply(reply="") == "暂时无法回复，请稍后再试。" + assert "API" in _user_facing_wechat_reply( + reply='Missing API key for provider "openai". Configure the gateway auth for that provider.' + ) + assert _user_facing_wechat_reply(reply="你好") == "你好" + + +def test_persist_channel_assistant_if_turn_missing_inserts_once() -> None: + class _Msg: + def __init__(self, role: str, turn_uuid: str) -> None: + self.role = role + self.turn_uuid = turn_uuid + + class _Store: + def __init__(self) -> None: + self.rows: list[_Msg] = [_Msg("user", "turn-1")] + self.added: list[tuple[str, str, str]] = [] + + def get_messages(self, *, session_id: str, limit: int = 80) -> list[_Msg]: + _ = (session_id, limit) + return list(self.rows) + + def add_message(self, **kwargs: object) -> None: + self.added.append( + ( + str(kwargs.get("session_id") or ""), + str(kwargs.get("turn_uuid") or ""), + str(kwargs.get("content") or ""), + ) + ) + self.rows.append(_Msg("assistant", str(kwargs.get("turn_uuid") or ""))) + + store = _Store() + _persist_channel_assistant_if_turn_missing( + store=store, + session_id="s1", + turn_uuid="", + final_text="暂时无法回复，请稍后再试。", + ) + assert len(store.added) == 1 + assert store.added[0] == ("s1", "turn-1", "暂时无法回复，请稍后再试。") + _persist_channel_assistant_if_turn_missing( + store=store, + session_id="s1", + turn_uuid="turn-1", + final_text="暂时无法回复，请稍后再试。", + ) + assert len(store.added) == 1 + + +def test_latest_user_turn_uuid() -> None: + class _Msg: + def __init__(self, role: str, turn_uuid: str) -> None: + self.role = role + self.turn_uuid = turn_uuid + + class _Store: + def get_messages(self, *, session_id: str, limit: int = 80) -> list[_Msg]: + _ = (session_id, limit) + return [_Msg("assistant", "a1"), _Msg("user", "u2")] + + assert _latest_user_turn_uuid(_Store(), session_id="s") == "u2" + + +def test_channel_dispatch_wechat_reads_weixin_settings() -> None: + class _Store: + def get_setting(self, key: str) -> str: + data = { + "channel.dispatch.interaction_mode.weixin": "comprehensive", + "channel.dispatch.specialist.weixin": "ops", + "channel.dispatch.lang.weixin": "zh", + } + return str(data.get(key) or "") + + mode, spec, lang = _resolve_channel_dispatch(_Store(), channel="wechat", account=None) + assert mode == "comprehensive" + assert spec == "ops" + assert lang == "zh" + assert _get_channel_dispatch_setting(_Store(), "channel.dispatch.specialist.", "wechat") == "ops" + + def test_parse_message_attachments_accepts_json_string() -> None: out = _parse_message_attachments('[{"type":"image_ref","attachment_id":"a1"}]') assert len(out) == 1 @@ -55,16 +142,38 @@ def test_collect_reply_attachments_prefers_matching_assistant_text() -> None: assert out[0].get("attachment_id") == "new" +def test_collect_reply_attachments_does_not_reuse_stale_images_on_text_only_reply() -> None: + rows = [ + _Row(role="assistant", content="here is a chart", attachments='[{"attachment_id":"old"}]'), + _Row(role="assistant", content="ok", attachments=None), + ] + out = _collect_reply_attachments_from_history(store=_FakeStore(rows), session_id="s1", reply_text="ok") + assert out == [] + + def test_collect_recent_tool_attachments_falls_back_to_tool_media() -> None: rows = [ - _Row(role="assistant", content="x", attachments=None), + _Row(role="user", content="draw", attachments=None), _Row(role="tool", content="{}", attachments='[{"type":"image_ref","attachment_id":"a1"}]'), + _Row(role="assistant", content="x", attachments=None), ] out = _collect_recent_tool_attachments(store=_FakeStore(rows), session_id="s1") assert len(out) == 1 assert out[0].get("attachment_id") == "a1" +def test_collect_recent_tool_attachments_ignores_media_from_prior_turn() -> None: + rows = [ + _Row(role="user", content="old question", attachments=None), + _Row(role="tool", content="{}", attachments='[{"type":"image_ref","attachment_id":"stale"}]'), + _Row(role="assistant", content="old answer", attachments=None), + _Row(role="user", content="new question", attachments=None), + _Row(role="assistant", content="new answer", attachments=None), + ] + out = _collect_recent_tool_attachments(store=_FakeStore(rows), session_id="s1") + assert out == [] + + def test_maybe_add_media_path_for_wechat_reply_sets_media_path(monkeypatch) -> None: # Avoid touching disk: stub AttachmentAssetStore.get_local_path. from pathlib import Path diff --git a/tests/test_pg_compat.py b/tests/test_pg_compat.py index fd6cfb10..95045984 100644 --- a/tests/test_pg_compat.py +++ b/tests/test_pg_compat.py @@ -44,3 +44,9 @@ def test_scrub_nul_bytes_from_text() -> None: def test_scrub_nul_bytes_from_jsonable_nested() -> None: assert pg_compat.scrub_nul_bytes_from_jsonable({"x": "y\x00z"}) == {"x": "yz"} + + +def test_escape_percent_in_sql_literals_for_like_b64_prefix() -> None: + sql = "SELECT 1 FROM llm_profile WHERE api_key LIKE 'b64:%'" + adapted = pg_compat.adapt_sql_for_postgres(sql) + assert "LIKE 'b64:%%'" in adapted diff --git a/tests/test_weixin_ilink_api.py b/tests/test_weixin_ilink_api.py index e17fb9e7..a200c853 100644 --- a/tests/test_weixin_ilink_api.py +++ b/tests/test_weixin_ilink_api.py @@ -1,5 +1,7 @@ from __future__ import annotations +import os +import time import unittest from fastapi.testclient import TestClient @@ -132,6 +134,36 @@ class WeixinIlinkApiTests(unittest.TestCase): finally: weixin_ilink_api._process_inbound_payload_usecase = old_usecase # type: ignore[assignment] + def test_native_reply_timeout_returns_user_message(self) -> None: + old_usecase = weixin_ilink_api._process_inbound_payload_usecase + old_timeout = weixin_ilink_api._native_reply_timeout_sec + + def _slow_usecase(payload: dict[str, object]) -> dict[str, object]: + _ = payload + time.sleep(1.2) + return {"ok": True, "replies": [{"text": "late"}]} + + try: + weixin_ilink_api._process_inbound_payload_usecase = _slow_usecase # type: ignore[assignment] + weixin_ilink_api._native_reply_timeout_sec = lambda: 0.2 # type: ignore[assignment] + r = self.client.post( + "/weixin/native/reply", + headers=self.headers, + json={ + "channel": "wechat", + "account_id": "bot-1", + "ctx": {"From": "wxid_u4", "To": "wxid_u4", "Body": "hi"}, + }, + ) + self.assertEqual(r.status_code, 200, r.text) + data = r.json() or {} + replies = data.get("replies") if isinstance(data.get("replies"), list) else [] + self.assertEqual(len(replies), 1, data) + self.assertIn("超时", str((replies[0] or {}).get("text") or "")) + finally: + weixin_ilink_api._process_inbound_payload_usecase = old_usecase # type: ignore[assignment] + weixin_ilink_api._native_reply_timeout_sec = old_timeout # type: ignore[assignment] + def test_whatsapp_native_reply_returns_sync_replies(self) -> None: old_usecase = weixin_ilink_api._process_inbound_payload_usecase