ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-13-G' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:57:35 -07:00
parent ac875db13e c299d5a33d
commit 7cfcc88a64
1 changed files with 74 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/platforms/signal.py
+74 -115
View File
@@ -1,9 +1,6 @@
"""Signal messenger platform adapter.
Connects to a signal-cli daemon in HTTP mode (``signal-cli daemon --http 127.0.0.1:8080``):
inbound messages arrive via SSE, outbound messages/actions use JSON-RPC 2.0 over HTTP.
Requires SIGNAL_HTTP_URL and SIGNAL_ACCOUNT.
"""
"""Signal messenger platform adapter for a signal-cli daemon in HTTP mode (``signal-cli daemon
--http 127.0.0.1:8080``): inbound via SSE, outbound via JSON-RPC 2.0. Requires SIGNAL_HTTP_URL
and SIGNAL_ACCOUNT."""
import asyncio
import base64
@@ -55,6 +52,11 @@ _MAGIC_EXTENSIONS = ((b"\x89PNG", ".png"), (b"\xff\xd8", ".jpg"), (b"GIF8", ".gi
_MEDIA_TYPE_BY_MIME_PREFIX = (
("audio/", MessageType.VOICE), ("image/", MessageType.PHOTO), ("video/", MessageType.VIDEO))
_OUTCOME_REACTION = {ProcessingOutcome.SUCCESS: "✅", ProcessingOutcome.FAILURE: "❌"}
# send_multiple_images skip reasons → logger.warning args (url, detail).
_SKIP_IMAGE_LOG = {
"download": lambda url, detail: ("Signal: failed to download image %s: %s", url, detail),
"missing": lambda url, detail: ("Signal: image file not found for %s", url),
"oversize": lambda url, detail: ("Signal: image too large (%d bytes), skipping %s", detail, url)}
_QUOTE_AUTHOR_KEYS = (
"author", "authorNumber", "authorUuid", "authorAci", "authorServiceId", "authorServiceIdString")
@@ -141,24 +143,20 @@ def _render_mentions(text: str, mentions: list) -> str:
def _is_signal_service_id(value: str) -> bool:
"""Return True if *value* already looks like a Signal service identifier."""
"""Return True if *value* already looks like a Signal service identifier (PNI:/u: prefix or UUID)."""
if not value:
return False
if value.startswith(("PNI:", "u:")):
return True
try:
with suppress(ValueError, AttributeError, TypeError):
uuid.UUID(value)
return True
except (ValueError, AttributeError, TypeError):
return False
return False
def _looks_like_e164_number(value: str) -> bool:
"""Return True for a plausible E.164 phone number."""
if not value or not value.startswith("+"):
return False
digits = value[1:]
return digits.isdigit() and 7 <= len(digits) <= 15
return bool(value) and value.startswith("+") and value[1:].isdigit() and 7 <= len(value) - 1 <= 15
def check_signal_requirements() -> bool:
@@ -189,15 +187,13 @@ class SignalAdapter(BasePlatformAdapter):
self.http_url = extra.get("http_url", "http://127.0.0.1:8080").rstrip("/")
self.account = extra.get("account", "")
self.ignore_stories = extra.get("ignore_stories", True)
# Allowlists are per-profile (scoped reads); group policy derives from the group
# allowlist's presence. Mention filter: only respond in groups when @mentioned.
# Allowlists are per-profile (scoped reads); group policy derives from the group allowlist's
# presence. The DM allowlist mirrors run.py's SIGNAL_ALLOWED_USERS so reaction hooks (which
# fire before run.py's auth gate) can skip unauthorized senders; "*" = open.
self.group_allow_from = set(_parse_comma_list(_sig_secret("SIGNAL_GROUP_ALLOWED_USERS", "")))
_rm_cfg = extra.get("require_mention")
self.require_mention = (
bool(_rm_cfg) if _rm_cfg is not None
else os.getenv("SIGNAL_REQUIRE_MENTION", "false").lower() in TRUTHY_STRINGS)
# DM allowlist mirrors run.py's SIGNAL_ALLOWED_USERS so reaction hooks (which fire
# before run.py's auth gate) can skip unauthorized senders. "*" = open.
self.require_mention = (bool(_rm_cfg) if _rm_cfg is not None
else os.getenv("SIGNAL_REQUIRE_MENTION", "false").lower() in TRUTHY_STRINGS)
self.dm_allow_from = set(_parse_comma_list(_sig_secret("SIGNAL_ALLOWED_USERS", "*")))
self.client: Optional[httpx.AsyncClient] = None
self._sse_task: Optional[asyncio.Task] = None
@@ -210,22 +206,21 @@ class SignalAdapter(BasePlatformAdapter):
self._last_sse_activity = 0.0
self._sse_response: Optional[httpx.Response] = None
self._account_normalized = self.account.strip()
# Recently sent timestamps filter echo-backs (Note to Self / linked-device sync-sents).
# LRU + TTL so a pending echo in a chatty group isn't evicted by many outbounds.
# Recently sent timestamps filter echo-backs (Note to Self / linked-device sync-sents); LRU + TTL
# so a pending echo in a chatty group isn't evicted by many outbounds.
self._recent_sent_timestamps: "OrderedDict[int, float]" = OrderedDict()
self._max_recent_timestamps = 512
self._recent_sent_ttl_seconds = 300.0
# Separate FIFO of outbound timestamps: Signal quote.id is the quoted message's
# timestamp, so replies to this bot are recognised after the echo was consumed.
# Separate FIFO of outbound timestamps: Signal quote.id is the quoted message's timestamp, so
# replies to this bot are recognised after the echo was consumed.
self._sent_message_timestamps: "OrderedDict[str, None]" = OrderedDict()
self._max_sent_message_timestamps = 500
# Best-effort number↔ACI/PNI UUID mapping so sends can upgrade a number to the UUID signal-cli prefers.
self._recipient_uuid_by_number: Dict[str, str] = {}
self._recipient_number_by_uuid: Dict[str, str] = {}
self._recipient_cache_lock = asyncio.Lock()
logger.info("Signal adapter initialized: url=%s account=%s groups=%s",
self.http_url, redact_phone(self.account),
"enabled" if self.group_allow_from else "disabled")
logger.info("Signal adapter initialized: url=%s account=%s groups=%s", self.http_url,
redact_phone(self.account), "enabled" if self.group_allow_from else "disabled")
async def connect(self, *, is_reconnect: bool = False) -> bool:
"""Connect to signal-cli daemon and start SSE listener."""
@@ -246,12 +241,12 @@ class SignalAdapter(BasePlatformAdapter):
try:
try:
resp = await self.client.get(f"{self.http_url}/api/v1/check", timeout=10.0)
if resp.status_code != 200:
logger.error("Signal: health check failed (status %d)", resp.status_code)
return False
except Exception as e:
logger.error("Signal: cannot reach signal-cli at %s: %s", self.http_url, e)
return False
if resp.status_code != 200:
logger.error("Signal: health check failed (status %d)", resp.status_code)
return False
self._running = True
self._last_sse_activity = time.time()
self._sse_task = asyncio.create_task(self._sse_listener())
@@ -266,11 +261,6 @@ class SignalAdapter(BasePlatformAdapter):
if lock_acquired:
self._release_platform_lock()
def _spawn_background(self, coro) -> None:
task = asyncio.create_task(coro)
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
async def _close_client(self) -> None:
if self.client:
await self.client.aclose()
@@ -375,7 +365,9 @@ class SignalAdapter(BasePlatformAdapter):
"""Force SSE reconnection by closing the current response."""
if self._sse_response and not self._sse_response.is_stream_consumed:
with suppress(Exception):
self._spawn_background(self._sse_response.aclose())
task = asyncio.create_task(self._sse_response.aclose())
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
self._sse_response = None
def _unwrap_sync_message(self, envelope_data: dict) -> Optional[dict]:
@@ -386,8 +378,7 @@ class SignalAdapter(BasePlatformAdapter):
if not sent_msg or not isinstance(sent_msg, dict):
return None
dest = sent_msg.get("destinationNumber") or sent_msg.get("destination")
sent_msg_group_id = (sent_msg.get("groupInfo") or {}).get("groupId")
if dest != self._account_normalized and not sent_msg_group_id:
if dest != self._account_normalized and not (sent_msg.get("groupInfo") or {}).get("groupId"):
return None
if self._consume_sent_timestamp(sent_msg.get("timestamp")):
return None # echo of our own outbound reply
@@ -442,9 +433,8 @@ class SignalAdapter(BasePlatformAdapter):
envelope_data = self._unwrap_sync_message(envelope_data)
if envelope_data is None:
return
sender = (
envelope_data.get("sourceNumber") or envelope_data.get("sourceUuid") or envelope_data.get("source")
)
sender = (envelope_data.get("sourceNumber") or envelope_data.get("sourceUuid")
or envelope_data.get("source"))
sender_name = envelope_data.get("sourceName", "")
sender_uuid = envelope_data.get("sourceUuid", "")
self._remember_recipient_identifiers(sender, sender_uuid)
@@ -457,9 +447,8 @@ class SignalAdapter(BasePlatformAdapter):
if self.ignore_stories and envelope_data.get("storyMessage"):
return
# Edited messages carry their updated dataMessage inside editMessage
data_message = (
envelope_data.get("dataMessage") or (envelope_data.get("editMessage") or {}).get("dataMessage")
)
data_message = (envelope_data.get("dataMessage")
or (envelope_data.get("editMessage") or {}).get("dataMessage"))
if not data_message:
return
group_info = data_message.get("groupInfo")
@@ -550,14 +539,10 @@ class SignalAdapter(BasePlatformAdapter):
return True
if not reply_to_author:
return False
author = str(reply_to_author).strip()
if self._account_normalized and author == self._account_normalized:
return True
cached_uuid = self._recipient_uuid_by_number.get(self._account_normalized)
if cached_uuid and author == cached_uuid:
return True
cached_number = self._recipient_number_by_uuid.get(author)
return bool(cached_number and cached_number == self._account_normalized)
author, acct = str(reply_to_author).strip(), self._account_normalized
# Cached number↔UUID mappings are only ever stored with truthy keys and values.
return bool(acct) and (author == acct or author == self._recipient_uuid_by_number.get(acct)
or self._recipient_number_by_uuid.get(author) == acct)
def _remember_sent_message_timestamp(self, timestamp: Any) -> None:
"""Keep a bounded cache of outbound Signal timestamps for quote matching."""
@@ -578,8 +563,8 @@ class SignalAdapter(BasePlatformAdapter):
profile = contact.get("profile")
if not service_id and isinstance(profile, dict):
service_id = profile.get("serviceId") or profile.get("uuid")
if service_id and _is_signal_service_id(service_id) and (
contact.get("number") == phone_number or contact.get("recipient") == phone_number):
if service_id and _is_signal_service_id(service_id) and phone_number in (
contact.get("number"), contact.get("recipient")):
return service_id
return None
@@ -595,12 +580,11 @@ class SignalAdapter(BasePlatformAdapter):
if cached:
return cached
contacts = await self._rpc("listContacts", {"account": self.account, "allRecipients": True})
if isinstance(contacts, list):
for contact in contacts:
number = contact.get("number") if isinstance(contact, dict) else None
service_id = self._extract_contact_uuid(contact, chat_id)
if number and service_id:
self._remember_recipient_identifiers(number, service_id)
for contact in contacts if isinstance(contacts, list) else ():
number = contact.get("number") if isinstance(contact, dict) else None
service_id = self._extract_contact_uuid(contact, chat_id)
if number and service_id:
self._remember_recipient_identifiers(number, service_id)
return self._recipient_uuid_by_number.get(chat_id, chat_id)
async def _with_target(self, params: Dict[str, Any], chat_id: str, *, resolve: bool = True) -> Dict[str, Any]:
@@ -616,8 +600,7 @@ class SignalAdapter(BasePlatformAdapter):
result = await self._rpc("getAttachment", {"account": self.account, "id": attachment_id})
if not result:
return None, ""
# signal-cli returns {"data": "base64..."}
if isinstance(result, dict):
if isinstance(result, dict): # signal-cli returns {"data": "base64..."}
result = result.get("data")
if not result:
logger.warning("Signal: attachment response missing 'data' key")
@@ -669,10 +652,8 @@ class SignalAdapter(BasePlatformAdapter):
logger.log(fail_level, "Signal RPC %s failed: %s", method, e)
return None
@staticmethod
def _markdown_to_signal(text: str) -> tuple[str, list[str]]:
"""Backward-compatible wrapper around shared Signal formatting helper."""
return markdown_to_signal(text)
# Backward-compatible alias for the shared formatting helper.
_markdown_to_signal = staticmethod(markdown_to_signal)
def format_message(self, content: str) -> str:
"""Plain-text fallback for the base-class send path; send() applies rich styles itself."""
@@ -680,9 +661,7 @@ class SignalAdapter(BasePlatformAdapter):
def _validate_send_result(self, result: Any) -> tuple[bool, Optional[str]]:
"""Validate signal-cli send response results. Returns (success, error_message)."""
if not result or not isinstance(result, dict):
return True, None
results = result.get("results")
results = result.get("results") if isinstance(result, dict) else None
for r in results if isinstance(results, list) else ():
if not isinstance(r, dict):
continue
@@ -709,26 +688,22 @@ class SignalAdapter(BasePlatformAdapter):
except (TypeError, ValueError):
logger.debug("[Signal] Ignoring malformed textStyle range: %r", style_string)
continue
overlap_start = max(style_start, chunk_start)
overlap_end = min(style_end, chunk_end)
overlap_start, overlap_end = max(style_start, chunk_start), min(style_end, chunk_end)
if overlap_start < overlap_end:
adjusted.append(f"{overlap_start - chunk_start}:{overlap_end - overlap_start}:{style_type}")
return adjusted
@classmethod
def _split_signal_formatted_message(
cls, plain_text: str, text_styles: list[str], max_length: int,
) -> list[tuple[str, list[str]]]:
cls, plain_text: str, text_styles: list[str], max_length: int) -> list[tuple[str, list[str]]]:
"""Split converted Signal text into chunks, translating body ranges per chunk. Splitting after
conversion keeps styles that cross a chunk boundary intact instead of leaking Markdown markers."""
if utf16_len(plain_text) <= max_length:
return [(plain_text, text_styles)]
indicator_reserve = 10 # Mirrors BasePlatformAdapter.truncate_message().
body_limit = max(1, max_length - indicator_reserve)
body_limit = max(1, max_length - 10) # 10 = indicator reserve, mirrors truncate_message().
offsets = cls._utf16_offsets(plain_text)
chunks: list[tuple[str, list[str]]] = []
start_idx = 0
total_u16 = offsets[-1]
start_idx, total_u16 = 0, offsets[-1]
while offsets[start_idx] < total_u16:
end_budget = min(total_u16, offsets[start_idx] + body_limit)
end_idx = start_idx + 1
@@ -753,9 +728,8 @@ class SignalAdapter(BasePlatformAdapter):
self._track_sent_timestamp(result)
return result, None
async def send(
self, chat_id: str, content: str, reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None) -> SendResult:
async def send(self, chat_id: str, content: str, reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None) -> SendResult:
"""Send a text message with native Signal formatting."""
await self._stop_typing_indicator(chat_id)
if not content or not content.strip():
@@ -770,8 +744,8 @@ class SignalAdapter(BasePlatformAdapter):
params["textStyle"] = text_styles[0]
elif text_styles:
params["textStyles"] = text_styles
logger.info("[Signal] Sending response chunk %d/%d (%d chars) to %s",
idx, len(chunks), len(plain_text), chat_id)
logger.info("[Signal] Sending response chunk %d/%d (%d chars) to %s", idx, len(chunks), len(plain_text),
chat_id)
last_result, err = await self._rpc_send(params, "RPC send failed")
if err:
return err
@@ -813,8 +787,7 @@ class SignalAdapter(BasePlatformAdapter):
self._typing_skip_until.pop(chat_id, None)
return
fails = self._typing_failures[chat_id] = fails + 1
# After 3 consecutive failures back off exponentially (16s, 32s, 60s cap).
if fails >= 3:
if fails >= 3: # exponential backoff: 16s, 32s, 60s cap
self._typing_skip_until[chat_id] = now + min(60.0, 16.0 * (2 ** (fails - 3)))
async def _resolve_image_path(self, image_url: str) -> Tuple[Optional[str], Optional[str], Any]:
@@ -852,18 +825,12 @@ class SignalAdapter(BasePlatformAdapter):
attachments.append(file_path)
continue
skipped[reason] += 1
if reason == "download":
logger.warning("Signal: failed to download image %s: %s", image_url, detail)
elif reason == "missing":
logger.warning("Signal: image file not found for %s", image_url)
else:
logger.warning("Signal: image too large (%d bytes), skipping %s", detail, image_url)
logger.warning(*_SKIP_IMAGE_LOG[reason](image_url, detail))
if not attachments:
logger.error("Signal: no valid images in batch of %d (download=%d missing=%d oversize=%d)",
len(images), skipped["download"], skipped["missing"], skipped["oversize"])
return
logger.info("Signal send_multiple_images: %d/%d images valid, sending in chunks",
len(attachments), len(images))
logger.info("Signal send_multiple_images: %d/%d images valid, sending in chunks", len(attachments), len(images))
base_params = await self._with_target({"account": self.account, "message": ""}, chat_id)
per = SIGNAL_MAX_ATTACHMENTS_PER_MSG
att_batches = [attachments[i:i + per] for i in range(0, len(attachments), per)]
@@ -880,8 +847,7 @@ class SignalAdapter(BasePlatformAdapter):
async def _send_attachment_batch(self, scheduler, params: Dict[str, Any], n: int, label: str) -> None:
"""Send one attachment batch with rate-limit pacing and a single transient retry. Tokens are
deducted only on validated success (None = server never accepted it); 429s feed the scheduler."""
send_timeout = _signal_send_timeout(n)
max_attempts = SIGNAL_RATE_LIMIT_MAX_ATTEMPTS
send_timeout, max_attempts = _signal_send_timeout(n), SIGNAL_RATE_LIMIT_MAX_ATTEMPTS
for attempt in range(1, max_attempts + 1):
await scheduler.acquire(n)
t0 = time.monotonic()
@@ -928,9 +894,9 @@ class SignalAdapter(BasePlatformAdapter):
if reason == "download":
logger.warning("Signal: failed to download image: %s", detail)
if reason:
error = {"download": str(detail), "missing": "Image file not found",
"oversize": f"Image too large ({detail} bytes)"}[reason]
return SendResult(success=False, error=error)
return SendResult(success=False, error={
"download": str(detail), "missing": "Image file not found",
"oversize": f"Image too large ({detail} bytes)"}[reason])
return await self._send_file(chat_id, file_path, caption, "RPC send with attachment failed")
async def _send_file(self, chat_id: str, file_path: str, caption: Optional[str], fail_error: str) -> SendResult:
@@ -940,9 +906,8 @@ class SignalAdapter(BasePlatformAdapter):
_, err = await self._rpc_send(params, fail_error)
return err or SendResult(success=True)
async def _send_attachment(
self, chat_id: str, file_path: str, media_label: str, caption: Optional[str] = None,
) -> SendResult:
async def _send_attachment(self, chat_id: str, file_path: str, media_label: str,
caption: Optional[str] = None) -> SendResult:
"""Send any local file as a Signal attachment (shared by send_document/image_file/voice/video)."""
await self._stop_typing_indicator(chat_id)
try:
@@ -991,8 +956,7 @@ class SignalAdapter(BasePlatformAdapter):
async def send_reaction(self, chat_id: str, emoji: str, target_author: str, target_timestamp: int) -> bool:
"""React to the message (author number/UUID, Signal ms timestamp) via signal-cli RPC."""
ok = await self._send_reaction_rpc(chat_id, {
"account": self.account, "emoji": emoji, "targetAuthor": target_author,
"targetTimestamp": target_timestamp})
"account": self.account, "emoji": emoji, "targetAuthor": target_author, "targetTimestamp": target_timestamp})
if not ok:
logger.debug("Signal: sendReaction failed (chat=%s, emoji=%s)", chat_id[:20], emoji)
return ok
@@ -1000,8 +964,8 @@ class SignalAdapter(BasePlatformAdapter):
async def remove_reaction(self, chat_id: str, target_author: str, target_timestamp: int) -> bool:
"""Remove a reaction by sending an empty-string emoji."""
return await self._send_reaction_rpc(chat_id, {
"account": self.account, "emoji": "", "targetAuthor": target_author,
"targetTimestamp": target_timestamp, "remove": True})
"account": self.account, "emoji": "", "targetAuthor": target_author, "targetTimestamp": target_timestamp,
"remove": True})
def _extract_reaction_target(self, event: MessageEvent) -> Optional[tuple]:
"""Extract (target_author, target_timestamp) from a MessageEvent, or None."""
@@ -1015,11 +979,8 @@ class SignalAdapter(BasePlatformAdapter):
so an unauthorized contact's 👀 would otherwise reveal a listening bot."""
if os.getenv("SIGNAL_REACTIONS", "true").lower() in {"false", "0", "no"}:
return False
if event is not None:
sender = getattr(getattr(event, "source", None), "user_id", None)
if sender and "*" not in self.dm_allow_from and sender not in self.dm_allow_from:
return False
return True
sender = getattr(getattr(event, "source", None), "user_id", None) if event is not None else None
return not (sender and "*" not in self.dm_allow_from and sender not in self.dm_allow_from)
async def on_processing_start(self, event: MessageEvent) -> None:
"""React with 👀 when processing begins."""
@@ -1036,11 +997,9 @@ class SignalAdapter(BasePlatformAdapter):
target = self._extract_reaction_target(event)
if not target:
return
chat_id = event.source.chat_id
await self.remove_reaction(chat_id, *target)
emoji = _OUTCOME_REACTION.get(outcome)
if emoji:
await self.send_reaction(chat_id, emoji, *target)
await self.remove_reaction(event.source.chat_id, *target)
if emoji := _OUTCOME_REACTION.get(outcome):
await self.send_reaction(event.source.chat_id, emoji, *target)
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
"""Get information about a chat/contact."""