hansjone/oclaw
1
0
Fork 0
mirror of https://github.com/hansjone/oclaw.git synced 2026-10-09 08:30:45 +08:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(whatsapp): deliver scheduled job attachments to groups

Browse source 
Scheduled outbound only sent text via pollOutboundQueue; collect deliverable attachments from the turn, encode them in source JSON, and send via sendReplyWithAttachments.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-07-21 14:29:55 +08:00
parent 40eed39110
commit f5cb64a3a4
7 changed files with 248 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

30
runtime/operations/whatsapp_bridge/baileys_runner.ts
View file

@ -897,16 +897,32 @@ async function pollOutboundQueue(sock: ReturnType<typeof makeWASocket>): 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",

73
runtime/scheduler/channel_delivery.py
View file

@ -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"}

8
runtime/scheduler/whatsapp_mentions.py
View file

@ -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)

1
runtime/scheduler/worker_turn.py
View file

@ -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(

26
svc/persistence/sqlite_store.py
View file

@ -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,

116
tests/test_whatsapp_attachment_delivery.py Normal file
View file

@ -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()

10
tests/test_whatsapp_ops_scripts.py
View file

@ -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