Broadcast bot: parity with official simplex-broadcast-bot

Relay publishers' text/links to all contacts via the native /feed command
(reports 'Forwarded to N contact(s), M errors'); reply to non-publishers with
the prohibited message and internally delete their message (CIDMInternal, as
upstream does). Filter content to text/links. Publishers accept 'Name' or
'ID:Name'; welcome/prohibited defaults list the publishers. Add publishers +
prohibited-reply fields to the create form. Adds broadcast_test.py (3 in-process
controllers: bot + publisher + subscriber) — passes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

manager/broadcast_test.py

@@ -0,0 +1,144 @@
+"""End-to-end test of the broadcast bot (Pattern 3, in-process FFI).
+
+Runs the real bot via profiles.start_bot, connects a publisher ("pub") and a
+non-publisher ("sub") to it, then checks:
+  - a publisher's message is broadcast to all contacts (sub receives it)
+  - a non-publisher's message gets the prohibited reply and is deleted
+
+Uses three libsimplex controllers in one process (bot + pub + sub) — exactly the
+multi-controller model the manager relies on. Needs network (SMP).
+
+Run:  .venv/bin/python broadcast_test.py
+"""
+
+import asyncio
+import json
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import profiles as pm  # noqa: E402
+from simplex_chat import ChatApi, SqliteDb  # noqa: E402
+
+DATA = Path("data")
+BOT_PREFIX = str(DATA / "bctest_bot")
+PUB_PREFIX = str(DATA / "bctest_pub")
+SUB_PREFIX = str(DATA / "bctest_sub")
+BOT_PID = 99001
+
+
+def cleanup():
+    for pat in ("bctest_bot_*", "bctest_pub_*", "bctest_sub_*"):
+        for p in DATA.glob(pat):
+            p.unlink()
+
+
+async def make_account(prefix: str, display: str) -> ChatApi:
+    chat = await ChatApi.init(SqliteDb(file_prefix=prefix))
+    user = await chat.api_get_active_user()
+    if not user:
+        await chat.api_create_active_user({"displayName": display, "fullName": ""})
+    await chat.start_chat()
+    return chat
+
+
+async def wait_until(fn, timeout=120, every=1):
+    start = time.time()
+    while time.time() - start < timeout:
+        v = await fn()
+        if v:
+            return v
+        await asyncio.sleep(every)
+    return None
+
+
+async def first_contact_id(chat: ChatApi) -> int | None:
+    u = await chat.api_get_active_user()
+    cs = await chat.api_list_contacts(u["userId"])
+    return cs[0]["contactId"] if cs else None
+
+
+async def incoming_texts(chat: ChatApi, contact_id: int) -> list[str]:
+    c = await chat.api_get_chat("direct", contact_id, 50)
+    out = []
+    for ci in c.get("chatItems", []):
+        d = ci.get("chatDir", {}).get("type", "")
+        if d.endswith("Rcv"):
+            out.append(ci.get("content", {}).get("msgContent", {}).get("text", ""))
+    return out
+
+
+async def main() -> int:
+    cleanup()
+    addr_box = {}
+
+    async def on_address(pid, addr):
+        addr_box["addr"] = addr
+
+    profile = {
+        "id": BOT_PID, "name": "bctestbot", "bot_type": "broadcast",
+        "db_prefix": BOT_PREFIX, "config": json.dumps({"publishers": ["pub"]}),
+    }
+    pub = sub = None
+    ok = True
+    try:
+        await pm.start_bot(profile, on_address)
+        addr = await wait_until(lambda: asyncio.sleep(0, addr_box.get("addr")), timeout=90)
+        print("bot address:", bool(addr))
+        assert addr, "bot never published an address"
+
+        pub = await make_account(PUB_PREFIX, "pub")
+        sub = await make_account(SUB_PREFIX, "sub")
+        await pub.send_chat_cmd(f"/connect {addr}")
+        await sub.send_chat_cmd(f"/connect {addr}")
+
+        pub_cid = await wait_until(lambda: first_contact_id(pub))
+        sub_cid = await wait_until(lambda: first_contact_id(sub))
+        print("pub connected:", bool(pub_cid), "| sub connected:", bool(sub_cid))
+        assert pub_cid and sub_cid, "publisher/subscriber did not connect"
+        # wait until the BOT itself has both contacts, else /feed would miss sub
+        both = await wait_until(
+            lambda: asyncio.sleep(0, len(pm.get_running(BOT_PID).contacts) >= 2), timeout=60
+        )
+        print("bot sees both contacts:", bool(both))
+
+        # 1) publisher broadcasts → sub should receive it
+        await pub.api_send_text_message({"chatType": "direct", "chatId": pub_cid}, "hello all")
+        got_bcast = await wait_until(
+            lambda: _contains(incoming_texts(sub, sub_cid), "hello all"), timeout=60, every=2
+        )
+        print("broadcast delivered to sub:", bool(got_bcast))
+        ok = ok and bool(got_bcast)
+
+        # 2) non-publisher (sub) sends → should get prohibited reply
+        await sub.api_send_text_message({"chatType": "direct", "chatId": sub_cid}, "spam please")
+        got_prohibited = await wait_until(
+            lambda: _contains(incoming_texts(sub, sub_cid), "deleted"), timeout=60, every=2
+        )
+        print("non-publisher got prohibited reply:", bool(got_prohibited))
+        ok = ok and bool(got_prohibited)
+    except AssertionError as e:
+        ok = False
+        print("ASSERT FAIL:", e)
+    finally:
+        await pm.stop_bot(BOT_PID)
+        for c in (pub, sub):
+            if c:
+                try:
+                    await c.close()
+                except Exception:
+                    pass
+        cleanup()
+
+    print("\nRESULT:", "PASS" if ok else "FAIL")
+    return 0 if ok else 1
+
+
+async def _contains(coro, needle):
+    texts = await coro
+    return any(needle.lower() in (t or "").lower() for t in texts)
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))

manager/profiles.py

@@ -323,6 +323,104 @@ async def delete_contact(profile_id: int, contact_id: int) -> bool:
     return True
 
 
+# ── Broadcast bot helpers (mirror the official simplex-broadcast-bot) ────────────
+# Publishers are configured as a list of "Name" or "ID:Name" strings. The official
+# bot matches a KnownContact by contactId AND display name; we match on either, so
+# names alone (what the UI collects) keep working.
+
+def _parse_publishers(pubs: list) -> tuple[set[int], set[str]]:
+    ids: set[int] = set()
+    names: set[str] = set()
+    for p in pubs or []:
+        s = str(p).strip()
+        if not s:
+            continue
+        if ":" in s:
+            left, right = s.split(":", 1)
+            if left.strip().isdigit():
+                ids.add(int(left.strip()))
+                names.add(right.strip())
+                continue
+        names.add(s)
+    return ids, names
+
+
+def _publisher_names(pubs: list) -> str:
+    _, names = _parse_publishers(pubs)
+    return ", ".join(sorted(names)) if names else "(no publishers configured)"
+
+
+def _bc_welcome(config: dict, name: str) -> str:
+    w = (config.get("welcome_message") or "").strip()
+    if w:
+        return w
+    return (
+        "Hello! I am a broadcast bot.\n"
+        f"I broadcast messages to all connected users from {_publisher_names(config.get('publishers', []))}."
+    )
+
+
+def _bc_prohibited(config: dict) -> str:
+    p = (config.get("prohibited_message") or "").strip()
+    if p:
+        return p
+    return (
+        f"Sorry, only these users can broadcast messages: {_publisher_names(config.get('publishers', []))}. "
+        "Your message is deleted."
+    )
+
+
+# Content types the broadcast bot will relay (matches the official allowlist).
+_BC_ALLOWED_CONTENT = {"text", "link"}
+
+
+async def _handle_broadcast_message(
+    b: "RunningBot", chat: Any, config: dict, item: dict, chat_info: dict, mc: dict, text: str
+) -> None:
+    """Mirror simplex-broadcast-bot: publishers' messages go to all contacts; everyone
+    else gets the prohibited reply and their message is deleted."""
+    ids, names = _parse_publishers(config.get("publishers", []))
+    ct = chat_info.get("contact", {})
+    sender_id = ct.get("contactId")
+    sender_name = ct.get("localDisplayName", "")
+    is_publisher = (sender_id in ids) or (sender_name in names)
+
+    async def reply(msg: str) -> None:
+        try:
+            await chat.api_send_text_reply(item, msg)
+        except Exception:
+            pass
+
+    if not is_publisher:
+        await reply(_bc_prohibited(config))
+        item_id = item.get("chatItem", {}).get("meta", {}).get("itemId")
+        if sender_id is not None and item_id is not None:
+            try:  # internal delete (a received message can't be deleted for the sender)
+                await chat.api_delete_chat_items("direct", sender_id, [item_id], "internal")
+            except Exception:
+                log.exception("broadcast: failed to delete non-publisher message")
+        return
+
+    if mc.get("type") not in _BC_ALLOWED_CONTENT or not text:
+        await reply("Message is not supported (text and links only).")
+        return
+
+    # Native broadcast: one /feed command fans out to every contact and reports counts.
+    try:
+        r = await chat.send_chat_cmd(f"/feed {text}")
+    except Exception as e:
+        log.error("broadcast /feed error: %s", e)
+        await reply("Could not broadcast right now, please try again.")
+        return
+    if isinstance(r, dict) and r.get("type") == "broadcastSent":
+        s, f = r.get("successes", 0), r.get("failures", 0)
+        _append_log(b, f"Broadcast → {s} ok, {f} errors")
+        await reply(f"Forwarded to {s} contact(s), {f} errors")
+    else:
+        log.error("broadcast unexpected response: %s", r)
+        await reply("Broadcast failed.")
+
+
 async def _run_bot(
     profile_id: int,
     name: str,
@@ -374,7 +472,10 @@ async def _run_bot(
             settings["businessAddress"] = True
             welcome = config.get("welcome_message", f"Welcome to {name} support.")
             settings["autoReply"] = {"type": "text", "text": welcome}
-        elif bot_type in ("echo", "broadcast", "directory", "deadmans"):
+        elif bot_type == "broadcast":
+            # auto-reply greets each new contact (default lists allowed publishers)
+            settings["autoReply"] = {"type": "text", "text": _bc_welcome(config, name)}
+        elif bot_type in ("echo", "directory", "deadmans"):
             welcome = config.get("welcome_message", f"Connected to {name}.")
             settings["autoReply"] = {"type": "text", "text": welcome}
 
@@ -437,16 +538,8 @@ async def _run_bot(
                 ct_name = ct.get("localDisplayName", "?")
                 _append_log(b, f"Contact connected: {ct_name}")
 
-                if bot_type == "echo":
-                    pass  # echo handled on message
-                elif bot_type == "broadcast":
-                    welcome = config.get("welcome_message", "You are subscribed.")
-                    try:
-                        await chat.api_send_text_message(
-                            {"chatType": "direct", "chatId": ct["contactId"]}, welcome
-                        )
-                    except Exception:
-                        pass
+                # echo replies on message; broadcast/others greet via the auto-reply
+                # configured in address settings, so nothing to do on connect here.
 
             elif tag == "newChatItems":
                 items = evt.get("chatItems", [])
@@ -517,18 +610,9 @@ async def _run_bot(
                                     pass
 
                         elif bot_type == "broadcast":
-                            publishers = config.get("publishers", [])
-                            sender = chat_info.get("contact", {}).get("localDisplayName", "")
-                            if sender in publishers and text:
-                                # broadcast to all contacts
-                                contacts = await chat.api_list_contacts(user_id)
-                                for c in contacts:
-                                    try:
-                                        await chat.api_send_text_message(
-                                            {"chatType": "direct", "chatId": c["contactId"]}, text
-                                        )
-                                    except Exception:
-                                        pass
+                            await _handle_broadcast_message(
+                                b, chat, config, item, chat_info, mc, text
+                            )
 
     except asyncio.CancelledError:
         pass

manager/templates/list.html

@@ -183,6 +183,22 @@
         <input type="text" name="superusers" placeholder="Alice, Bob">
       </div>
     </div>
+    <div id="broadcast-fields" style="display:none;">
+      <div style="border-top:1px solid var(--border);margin:4px 0 14px;padding-top:14px;">
+        <p class="muted" style="margin-bottom:12px;">
+          Only listed publishers can broadcast; their text/links are relayed to every contact.
+          Anyone else gets the prohibited reply and their message is deleted.
+        </p>
+      </div>
+      <div class="field">
+        <label>Publishers <span class="muted" style="font-weight:400;">(comma-separated; "Name" or "ID:Name")</span></label>
+        <input type="text" name="publishers" placeholder="Alice, 2:Bob">
+      </div>
+      <div class="field">
+        <label>Prohibited reply <span class="muted" style="font-weight:400;">(blank = default listing publishers)</span></label>
+        <input type="text" name="prohibited_message" placeholder="Only publishers can broadcast. Your message is deleted.">
+      </div>
+    </div>
     {% endif %}
     <div class="flex gap-8 mt-16" style="justify-content:flex-end;">
       <button type="button" class="btn btn-ghost"
@@ -255,6 +271,7 @@ function onTypeChange() {
   document.getElementById('support-fields').style.display = (val === 'support') ? 'block' : 'none';
   document.getElementById('deadmans-fields').style.display = (val === 'deadmans') ? 'block' : 'none';
   document.getElementById('directory-fields').style.display = (val === 'directory') ? 'block' : 'none';
+  document.getElementById('broadcast-fields').style.display = (val === 'broadcast') ? 'block' : 'none';
 }
 {% endif %}
 
@@ -293,6 +310,12 @@ document.getElementById('create-form').addEventListener('submit', async (e) => {
     const su = (fd.get('superusers') || '').split(',').map(s => s.trim()).filter(Boolean);
     if (su.length) config.superusers = su;
   }
+  if (botType === 'broadcast') {
+    const pubs = (fd.get('publishers') || '').split(',').map(s => s.trim()).filter(Boolean);
+    if (pubs.length) config.publishers = pubs;
+    const prohibited = (fd.get('prohibited_message') || '').trim();
+    if (prohibited) config.prohibited_message = prohibited;
+  }
   {% endif %}
   // Shared profile fields (users and bots)
   const bio = (fd.get('bio') || '').trim();