iarv/Remote-Terminal-for-MeshCore
1
0
Fork 1
mirror of https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git synced 2026-03-28 17:43:05 +01:00
Code Issues Packages Projects Releases Wiki Activity

Support incoming/outgoing detection in bots

Browse Source
This commit is contained in:
Jack Kingsman
2026-02-12 23:52:49 -08:00
parent 0fcf6a5653
commit 430b5aaba7
3 changed files with 167 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
app/bot.py
View File

@@ -11,6 +11,7 @@ the security implications.
"""
import asyncio
import inspect
import logging
import time
from concurrent.futures import ThreadPoolExecutor
@@ -48,15 +49,19 @@ def execute_bot_code(
channel_name: str | None,
sender_timestamp: int | None,
path: str | None,
is_outgoing: bool = False,
) -> str | list[str] | None:
"""
Execute user-provided bot code with message context.
The code should define a function:
`bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path)`
`bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing)`
that returns either None (no response), a string (single response message),
or a list of strings (multiple messages sent in order).
Legacy bot functions with 8 parameters (without is_outgoing) are detected
via inspect and called without the new parameter for backward compatibility.
Args:
code: Python code defining the bot function
sender_name: Display name of the sender (may be None)
@@ -67,6 +72,7 @@ def execute_bot_code(
channel_name: Channel name (e.g. "#general" with hash), None for DMs
sender_timestamp: Sender's timestamp from the message (may be None)
path: Hex-encoded routing path (may be None)
is_outgoing: True if this is our own outgoing message
Returns:
Response string, list of strings, or None.
@@ -95,18 +101,63 @@ def execute_bot_code(
bot_func = namespace["bot"]
# Detect whether the bot function accepts is_outgoing (new 9-param signature)
# or uses the legacy 8-param signature, for backward compatibility.
# Three cases: explicit is_outgoing param or 9+ params (positional),
# **kwargs (pass as keyword), or legacy 8-param (omit).
call_style = "legacy" # "positional", "keyword", or "legacy"
try:
# Call the bot function with message context
result = bot_func(
sender_name,
sender_key,
message_text,
is_dm,
channel_key,
channel_name,
sender_timestamp,
path,
)
sig = inspect.signature(bot_func)
params = sig.parameters
non_variadic = [
p
for p in params.values()
if p.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
]
if "is_outgoing" in params or len(non_variadic) >= 9:
call_style = "positional"
elif any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()):
call_style = "keyword"
except (ValueError, TypeError):
pass
try:
# Call the bot function with appropriate signature
if call_style == "positional":
result = bot_func(
sender_name,
sender_key,
message_text,
is_dm,
channel_key,
channel_name,
sender_timestamp,
path,
is_outgoing,
)
elif call_style == "keyword":
result = bot_func(
sender_name,
sender_key,
message_text,
is_dm,
channel_key,
channel_name,
sender_timestamp,
path,
is_outgoing=is_outgoing,
)
else:
result = bot_func(
sender_name,
sender_key,
message_text,
is_dm,
channel_key,
channel_name,
sender_timestamp,
path,
)
# Validate result
if result is None:
@@ -281,6 +332,7 @@ async def run_bot_for_message(
channel_name,
sender_timestamp,
path,
is_outgoing,
),
timeout=BOT_EXECUTION_TIMEOUT,
)

8
frontend/src/components/SettingsModal.tsx
View File

@@ -163,9 +163,10 @@ export function SettingsModal(props: SettingsModalProps) {
channel_name: str | None,
sender_timestamp: int | None,
path: str | None,
is_outgoing: bool = False,
) -> str | list[str] | None:
"""
Process incoming messages and optionally return a reply.
Process messages and optionally return a reply.
Args:
sender_name: Display name of sender (may be None)
@@ -176,11 +177,16 @@ export function SettingsModal(props: SettingsModalProps) {
channel_name: Channel name with hash (e.g. "#bot"), None for DMs
sender_timestamp: Sender's timestamp (unix seconds, may be None)
path: Hex-encoded routing path (may be None)
is_outgoing: True if this is our own outgoing message
Returns:
None for no reply, a string for a single reply,
or a list of strings to send multiple messages in order
"""
# Don't reply to our own outgoing messages
if is_outgoing:
return None
# Example: Only respond in #bot channel to "!pling" command
if channel_name == "#bot" and "!pling" in message_text.lower():
return "[BOT] Plong!"

110
tests/test_bot.py
View File

@@ -22,7 +22,7 @@ class TestExecuteBotCode:
def test_valid_code_returning_string(self):
"""Bot code that returns a string works correctly."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return f"Hello, {sender_name}!"
"""
result = execute_bot_code(
@@ -41,7 +41,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
def test_valid_code_returning_none(self):
"""Bot code that returns None works correctly."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return None
"""
result = execute_bot_code(
@@ -60,7 +60,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
def test_empty_string_response_treated_as_none(self):
"""Bot returning empty/whitespace string is treated as None."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return " "
"""
result = execute_bot_code(
@@ -135,7 +135,7 @@ bot = "I'm a string, not a function"
def test_bot_function_raises_exception(self):
"""Bot function that raises exception returns None."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
raise ValueError("oops!")
"""
result = execute_bot_code(
@@ -154,7 +154,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
def test_bot_returns_non_string(self):
"""Bot function returning non-string returns None."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return 42
"""
result = execute_bot_code(
@@ -201,10 +201,9 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
assert result is None
def test_bot_receives_all_parameters(self):
"""Bot function receives all expected parameters."""
"""Bot function receives all expected parameters including is_outgoing."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
# Verify all params are accessible
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
parts = [
f"name={sender_name}",
f"key={sender_key}",
@@ -214,6 +213,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
f"ch_name={channel_name}",
f"ts={sender_timestamp}",
f"path={path}",
f"outgoing={is_outgoing}",
]
return "|".join(parts)
"""
@@ -227,16 +227,98 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
channel_name="#test",
sender_timestamp=12345,
path="001122",
is_outgoing=True,
)
assert (
result
== "name=Bob|key=def456|msg=Test|dm=False|ch_key=AABBCCDD|ch_name=#test|ts=12345|path=001122"
== "name=Bob|key=def456|msg=Test|dm=False|ch_key=AABBCCDD|ch_name=#test|ts=12345|path=001122|outgoing=True"
)
def test_is_outgoing_defaults_to_false(self):
"""is_outgoing defaults to False when not specified."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return f"outgoing={is_outgoing}"
"""
result = execute_bot_code(
code=code,
sender_name="Alice",
sender_key="abc123",
message_text="Hi",
is_dm=True,
channel_key=None,
channel_name=None,
sender_timestamp=None,
path=None,
)
assert result == "outgoing=False"
def test_is_outgoing_true_passed_correctly(self):
"""is_outgoing=True is passed to bot function."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
if is_outgoing:
return None
return "reply"
"""
result = execute_bot_code(
code=code,
sender_name="Alice",
sender_key="abc123",
message_text="Hi",
is_dm=True,
channel_key=None,
channel_name=None,
sender_timestamp=None,
path=None,
is_outgoing=True,
)
assert result is None
def test_legacy_8_param_bot_still_works(self):
"""Legacy bot with 8 parameters (no is_outgoing) still works."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
return f"Hello, {sender_name}!"
"""
result = execute_bot_code(
code=code,
sender_name="Alice",
sender_key="abc123",
message_text="Hi",
is_dm=True,
channel_key=None,
channel_name=None,
sender_timestamp=None,
path=None,
is_outgoing=True,
)
assert result == "Hello, Alice!"
def test_legacy_bot_with_kwargs_receives_is_outgoing(self):
"""Legacy bot using **kwargs receives is_outgoing."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, **kwargs):
return f"outgoing={kwargs.get('is_outgoing', 'missing')}"
"""
result = execute_bot_code(
code=code,
sender_name="Alice",
sender_key="abc123",
message_text="Hi",
is_dm=True,
channel_key=None,
channel_name=None,
sender_timestamp=None,
path=None,
is_outgoing=True,
)
assert result == "outgoing=True"
def test_channel_message_with_none_sender_key(self):
"""Channel messages correctly pass None for sender_key."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
if sender_key is None and not is_dm:
return "channel message detected"
return "unexpected"
@@ -257,7 +339,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
def test_bot_returns_list_of_strings(self):
"""Bot function returning list of strings works correctly."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return ["First message", "Second message", "Third message"]
"""
result = execute_bot_code(
@@ -276,7 +358,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
def test_bot_returns_empty_list(self):
"""Bot function returning empty list is treated as None."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return []
"""
result = execute_bot_code(
@@ -295,7 +377,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
def test_bot_returns_list_with_empty_strings_filtered(self):
"""Bot function returning list filters out empty/whitespace strings."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return ["Valid", "", " ", "Also valid", None, 42]
"""
result = execute_bot_code(
@@ -315,7 +397,7 @@ def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name,
def test_bot_returns_list_all_empty_treated_as_none(self):
"""Bot function returning list of all empty strings is treated as None."""
code = """
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path):
def bot(sender_name, sender_key, message_text, is_dm, channel_key, channel_name, sender_timestamp, path, is_outgoing):
return ["", " ", ""]
"""
result = execute_bot_code(