diff --git a/runtime/operations/whatsapp_bridge/baileys_runner.ts b/runtime/operations/whatsapp_bridge/baileys_runner.ts index a375bd68..4dc9e9d9 100644 --- a/runtime/operations/whatsapp_bridge/baileys_runner.ts +++ b/runtime/operations/whatsapp_bridge/baileys_runner.ts @@ -897,16 +897,32 @@ async function pollOutboundQueue(sock: ReturnType): Promise const chatId = String((item as any).chat_id || "").trim(); const text = String((item as any).text || "").trim(); const source = String((item as any).source || "").trim(); - if (!id || !chatId || !text) continue; + const attachments = Array.isArray((item as any).attachments) ? ((item as any).attachments as Json[]) : []; + const mediaPath = String((item as any).media_path || (item as any).mediaPath || "").trim(); + const hasMedia = attachments.length > 0 || Boolean(mediaPath); + if (!id || !chatId || (!text && !hasMedia)) continue; let ok = true; let err = ""; try { - const sendContent = buildOutboundSendContent(text, source, sock, chatId); - const sent = await sock.sendMessage(chatId, sendContent); - const stanzaId = String((sent as any)?.key?.id || "").trim(); - log( - `outbound sent id=${id} chat=${chatId} stanza=${stanzaId || "?"} mentions=${(sendContent.mentions || []).length} mention0=${(sendContent.mentions || [])[0] || ""} text=${sendContent.text.slice(0, 80)}`, - ); + let stanzaId = ""; + if (hasMedia) { + const reply: Json = { + attachments, + metadata: decodeOutboundSource(source), + ...(mediaPath ? { media_path: mediaPath } : {}), + }; + await sendReplyWithAttachments({ sock, deliverTo: chatId, text, reply }); + log( + `outbound sent id=${id} chat=${chatId} attachments=${attachments.length} mediaPath=${mediaPath ? "yes" : "no"} text=${text.slice(0, 80)}`, + ); + } else { + const sendContent = buildOutboundSendContent(text, source, sock, chatId); + const sent = await sock.sendMessage(chatId, sendContent); + stanzaId = String((sent as any)?.key?.id || "").trim(); + log( + `outbound sent id=${id} chat=${chatId} stanza=${stanzaId || "?"} mentions=${(sendContent.mentions || []).length} mention0=${(sendContent.mentions || [])[0] || ""} text=${sendContent.text.slice(0, 80)}`, + ); + } try { await fetch(`${LOCAL_BASE_URL.replace(/\/+$/, "")}/whatsapp/outbound/ack`, { method: "POST", diff --git a/runtime/scheduler/channel_delivery.py b/runtime/scheduler/channel_delivery.py index 1b3fbd37..780860f6 100644 --- a/runtime/scheduler/channel_delivery.py +++ b/runtime/scheduler/channel_delivery.py @@ -4,7 +4,7 @@ import json import os from typing import Any -from runtime.orchestration.group_ingest import is_nonsend_channel_reply_text, should_send_channel_reply_text +from runtime.orchestration.group_ingest import is_nonsend_channel_reply_text from runtime.scheduler.session_resolver import parse_delivery_json from runtime.scheduler.whatsapp_mentions import ( encode_whatsapp_outbound_source, @@ -175,6 +175,55 @@ def persist_channel_context_token( ) +def _collect_scheduled_turn_attachments( + *, + store: Any, + session_id: str, + turn_uuid: str, +) -> list[dict[str, Any]]: + from runtime.application.gateway.inbound_service import ( + _is_channel_deliverable_attachment, + _parse_message_attachments, + ) + + sid = str(session_id or "").strip() + tu = str(turn_uuid or "").strip() + if not sid: + return [] + try: + rows = store.get_messages(session_id=sid, limit=120) + except Exception: + rows = [] + deliverable: list[dict[str, Any]] = [] + for row in reversed(list(rows or [])): + role = str(getattr(row, "role", "") or "").strip().lower() + if role != "tool": + continue + if tu and str(getattr(row, "turn_uuid", "") or "").strip() != tu: + continue + for att in _parse_message_attachments(getattr(row, "attachments", None)): + if _is_channel_deliverable_attachment(att): + deliverable.append(att) + return deliverable + + +def _prepare_channel_outbound_attachments( + attachments: list[dict[str, Any]], +) -> tuple[list[dict[str, Any]], str]: + if not attachments: + return [], "" + reply: dict[str, Any] = {"attachments": list(attachments)} + from runtime.application.gateway.inbound_service import ( + _maybe_add_media_path_for_wechat_reply, + _maybe_expand_reply_attachments_for_channel, + ) + + _maybe_expand_reply_attachments_for_channel(reply) + _maybe_add_media_path_for_wechat_reply(reply) + out_atts = [a for a in (reply.get("attachments") or []) if isinstance(a, dict)] + return out_atts, str(reply.get("media_path") or "").strip() + + def extract_context_token_from_inbound_metadata(metadata: dict[str, Any] | None) -> str: meta = metadata if isinstance(metadata, dict) else {} raw = meta.get("raw") if isinstance(meta.get("raw"), dict) else {} @@ -201,9 +250,20 @@ def deliver_scheduled_reply( resolved_chat_id: str = "", resolved_account_id: str = "", session_id: str = "", + turn_uuid: str = "", ) -> dict[str, Any]: text = str(reply_text or "").strip() - if not text or is_nonsend_channel_reply_text(text): + if is_nonsend_channel_reply_text(text): + text = "" + outbound_attachments, media_path = _prepare_channel_outbound_attachments( + _collect_scheduled_turn_attachments( + store=store, + session_id=session_id, + turn_uuid=turn_uuid, + ) + ) + has_attachments = bool(outbound_attachments or media_path) + if not text and not has_attachments: return {"ok": False, "skipped": True, "reason": "empty_or_silent_reply"} ch_lower = str(resolved_channel or "").strip().lower() @@ -221,7 +281,7 @@ def deliver_scheduled_reply( account_id = str( wa.get("account_id") or resolved_account_id or os.getenv("AIA_WHATSAPP_ACCOUNT_ID") or "wa-default" ).strip() - if wa_enabled and chat_id and should_send_channel_reply_text(text): + if wa_enabled and chat_id and (text or has_attachments): mention_jids = normalize_whatsapp_mention_jids(wa.get("mention_jids")) if not mention_jids: mention_jids = infer_whatsapp_mention_jids_from_text( @@ -264,6 +324,8 @@ def deliver_scheduled_reply( mention_jids=mention_jids, mention_names=mention_names, mention_text_ready=bool(mention_jids), + attachments=outbound_attachments if has_attachments else None, + media_path=media_path or None, ), ) results["whatsapp"] = { @@ -271,11 +333,12 @@ def deliver_scheduled_reply( "message_id": msg_id, "chat_id": chat_id, "mention_jids": mention_jids, + "attachments": len(outbound_attachments) if has_attachments else 0, } wx = delivery.get("weixin") if isinstance(delivery.get("weixin"), dict) else {} wx_enabled = bool(wx.get("enabled", True)) - if wx_enabled and should_send_channel_reply_text(text): + if wx_enabled and (text or has_attachments): from runtime.scheduler.weixin_delivery import resolve_weixin_delivery_target target = resolve_weixin_delivery_target( @@ -300,6 +363,8 @@ def deliver_scheduled_reply( context_token=context_token, store=store, tenant_id=tenant_id, + attachments=outbound_attachments if has_attachments else None, + media_path=media_path or None, ) else: results["weixin"] = {"ok": False, "error": "weixin_chat_missing"} diff --git a/runtime/scheduler/whatsapp_mentions.py b/runtime/scheduler/whatsapp_mentions.py index 414ff05e..8acd7d90 100644 --- a/runtime/scheduler/whatsapp_mentions.py +++ b/runtime/scheduler/whatsapp_mentions.py @@ -14,6 +14,8 @@ def encode_whatsapp_outbound_source( mention_jids: list[str] | None = None, mention_names: list[str] | None = None, mention_text_ready: bool = False, + attachments: list[dict[str, Any]] | None = None, + media_path: str | None = None, ) -> str: payload: dict[str, Any] = {"kind": str(kind or "scheduled_job")} jids = normalize_whatsapp_mention_jids(mention_jids) @@ -24,6 +26,12 @@ def encode_whatsapp_outbound_source( payload["mention_names"] = names if mention_text_ready: payload["mention_text_ready"] = True + atts = [a for a in (attachments or []) if isinstance(a, dict)] + if atts: + payload["attachments"] = atts + mp = str(media_path or "").strip() + if mp: + payload["media_path"] = mp return json.dumps(payload, ensure_ascii=False) diff --git a/runtime/scheduler/worker_turn.py b/runtime/scheduler/worker_turn.py index 779355db..0ac37c8e 100644 --- a/runtime/scheduler/worker_turn.py +++ b/runtime/scheduler/worker_turn.py @@ -128,6 +128,7 @@ def finalize_scheduled_turn_success( resolved_chat_id=str(payload.get("resolved_chat_id") or ""), resolved_account_id=str(payload.get("resolved_account_id") or ""), session_id=str(payload.get("session_id") or ""), + turn_uuid=str(base_result.get("turn_uuid") or payload.get("run_id") or ""), ) if scheduled_run_id: store.scheduled_job_run_update( diff --git a/svc/persistence/sqlite_store.py b/svc/persistence/sqlite_store.py index 32bd61c7..7f2172bb 100644 --- a/svc/persistence/sqlite_store.py +++ b/svc/persistence/sqlite_store.py @@ -6436,19 +6436,35 @@ class SqliteStore(ScheduledJobStoreMixin): """, (str(channel), str(account_id), utc_now_iso(), lim), ).fetchall() - return [ - { + out: list[dict[str, Any]] = [] + for r in rows: + source = str(r["source"] or "") + meta: dict[str, Any] = {} + if source.startswith("{"): + try: + parsed = json.loads(source) + if isinstance(parsed, dict): + meta = parsed + except Exception: + meta = {} + item: dict[str, Any] = { "id": str(r["id"] or ""), "tenant_id": str(r["tenant_id"] or ""), "channel": str(r["channel"] or ""), "account_id": str(r["account_id"] or ""), "chat_id": str(r["chat_id"] or ""), "text": str(r["text"] or ""), - "source": str(r["source"] or ""), + "source": source, "created_at": str(r["created_at"] or ""), } - for r in rows - ] + atts = [a for a in (meta.get("attachments") or []) if isinstance(a, dict)] + if atts: + item["attachments"] = atts + mp = str(meta.get("media_path") or "").strip() + if mp: + item["media_path"] = mp + out.append(item) + return out def list_pending_weixin_outbound_messages( self, diff --git a/tests/test_whatsapp_attachment_delivery.py b/tests/test_whatsapp_attachment_delivery.py new file mode 100644 index 00000000..c6445e67 --- /dev/null +++ b/tests/test_whatsapp_attachment_delivery.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from runtime.scheduler.channel_delivery import deliver_scheduled_reply +from runtime.scheduler.whatsapp_mentions import encode_whatsapp_outbound_source +from svc.persistence.sqlite_store import SqliteStore + + +class WhatsappAttachmentDeliveryTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db = Path(self._tmp.name) / "wa_attach.sqlite" + self.store = SqliteStore(str(self.db)) + tenant = self.store.create_tenant("Team") + self.tenant_id = str(tenant["id"]) + sess = self.store.create_session("WA scheduled") + self.session_id = str(sess.id) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_encode_whatsapp_outbound_source_with_attachments(self) -> None: + atts = [{"name": "report.xlsx", "mime": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "data_base64": "abc"}] + raw = encode_whatsapp_outbound_source( + mention_jids=["111@lid"], + attachments=atts, + media_path="/tmp/report.xlsx", + ) + data = json.loads(raw) + self.assertEqual(data.get("mention_jids"), ["111@lid"]) + self.assertEqual(data.get("attachments"), atts) + self.assertEqual(data.get("media_path"), "/tmp/report.xlsx") + + def test_list_pending_whatsapp_outbound_includes_attachments_from_source(self) -> None: + source = encode_whatsapp_outbound_source( + attachments=[{"name": "a.txt", "data_base64": "dGVzdA=="}], + media_path="D:/tmp/a.txt", + ) + self.store.enqueue_channel_outbound_message( + channel="whatsapp", + chat_id="120363012345678@g.us", + text="file attached", + tenant_id=self.tenant_id, + account_id="wa-default", + source=source, + ) + items = self.store.list_pending_channel_outbound_messages(channel="whatsapp", account_id="wa-default", limit=5) + self.assertEqual(len(items), 1) + self.assertEqual(len(items[0].get("attachments") or []), 1) + self.assertEqual(items[0].get("media_path"), "D:/tmp/a.txt") + + def test_deliver_scheduled_reply_encodes_tool_attachments_for_turn(self) -> None: + turn_uuid = "turn-wa-attach-1" + self.store.add_message( + session_id=self.session_id, + role="tool", + content='{"ok": true}', + turn_uuid=turn_uuid, + event_type="tool_result", + attachments=json.dumps( + [ + { + "type": "binary_ref", + "attachment_id": "att-1", + "name": "daily.xlsx", + "mime": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "deliverable": True, + } + ], + ensure_ascii=False, + ), + ) + delivery = { + "whatsapp": { + "enabled": True, + "target_type": "group", + "chat_id": "120363012345678@g.us", + "account_id": "wa-default", + }, + "weixin": {"enabled": False}, + } + with mock.patch( + "runtime.scheduler.channel_delivery._prepare_channel_outbound_attachments", + return_value=( + [{"name": "daily.xlsx", "mime": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "data_base64": "ZGF0YQ=="}], + "", + ), + ): + result = deliver_scheduled_reply( + self.store, + tenant_id=self.tenant_id, + reply_text="今日报表已生成", + delivery_json=json.dumps(delivery, ensure_ascii=False), + session_id=self.session_id, + turn_uuid=turn_uuid, + ) + self.assertTrue(result.get("ok"), result) + pending = self.store.list_pending_channel_outbound_messages( + channel="whatsapp", + account_id="wa-default", + limit=5, + ) + self.assertEqual(len(pending), 1) + source = json.loads(str(pending[0].get("source") or "{}")) + self.assertEqual(len(source.get("attachments") or []), 1) + self.assertEqual(source["attachments"][0].get("name"), "daily.xlsx") + self.assertEqual((result.get("channels") or {}).get("whatsapp", {}).get("attachments"), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_whatsapp_ops_scripts.py b/tests/test_whatsapp_ops_scripts.py index e7d33e34..88d29f44 100644 --- a/tests/test_whatsapp_ops_scripts.py +++ b/tests/test_whatsapp_ops_scripts.py @@ -39,6 +39,16 @@ def test_start_all_gracefully_skips_missing_channel_sidecars() -> None: assert 'Warn "whatsapp sidecar skipped:' in text +def test_whatsapp_runner_outbound_poll_uses_reply_attachments() -> None: + text = _read("runtime/operations/whatsapp_bridge/baileys_runner.ts") + assert "sendReplyWithAttachments" in text + assert "hasMedia" in text + assert "pollOutboundQueue" in text + poll_start = text.index("async function pollOutboundQueue") + poll_chunk = text[poll_start : poll_start + 2500] + assert "sendReplyWithAttachments" in poll_chunk + + def test_whatsapp_runner_supports_reply_attachments_base64() -> None: text = _read("runtime/operations/whatsapp_bridge/baileys_runner.ts") assert "sendReplyWithAttachments" in text