"""Inbound Hook — OpenClaw message gateway for the Intent Router.
This module is the bridge between OpenClaw's Telegram message flow and the
family assistant's intent engine. It runs on every inbound group message,
checks trigger conditions, and routes qualifying messages to the intent
router for calendar mutations.
TRIGGER CONDITIONS (strictly enforced):
1. Message contains a wake word ("Socrates", "calendar", "schedule", or
any word in INTENT_WAKE_WORDS env var)
2. Message is a direct reply to one of the bot's previous messages
If neither condition is met, the hook returns silently. No passive listening.
No processing of general family chatter.
Usage from OpenClaw agent context:
from icarus.core.inbound_hook import process_inbound
result = process_inbound(message, is_reply_to_bot, quoted_text)
if result and result.get("notification"):
# Send notification back to the group
"""
import json
import os
import re
from icarus.core.intent_router import should_process, process_group_message
from icarus.core.hermes import push_pipeline_results
Family group chat ID — used to route notifications back
FAMILY_GROUP_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
Wake words from env or defaults (same as intent_router)
WAKE_WORDS = os.environ.get(
"INTENT_WAKE_WORDS", "Socrates,calendar,schedule,@HomeButler"
).split(",")
def check_trigger(message: str, is_reply_to_bot: bool = False) -> bool:
"""Check if an inbound message meets trigger conditions.
This is a public wrapper around intent_router.should_process that
also handles @mention detection.
Args:
message: The raw text of the inbound Telegram message
is_reply_to_bot: True if the message is a reply to a bot message
Returns:
True if the message should be routed through the intent engine
"""
if is_reply_to_bot:
return True
message_lower = message.lower()
# Check wake words (including @mentions)
for word in WAKE_WORDS:
word = word.strip().lower()
if word and word in message_lower:
return True
return False
def strip_wake_word(message: str) -> str:
"""Remove the wake word from the message before passing to the intent engine.
The LLM doesn't need "Socrates" or "@HomeButler" in the message text —
it adds noise to intent extraction.
Args:
message: The raw message text
Returns:
The message with wake words stripped
"""
cleaned = message
for word in WAKE_WORDS:
word = word.strip()
if word:
# Case-insensitive replacement
cleaned = re.sub(re.escape(word), "", cleaned, flags=re.IGNORECASE)
# Clean up extra whitespace left behind
cleaned = re.sub(r"\s+", " ", cleaned).strip()
# Remove leading comma/period if wake word was at start
cleaned = re.sub(r"^[,.]\s*", "", cleaned)
return cleaned
def process_inbound(
message: str,
sender: str = "",
is_reply_to_bot: bool = False,
quoted_text: str = "",
dry_run: bool = False,
) -> dict | None:
"""Main entry point for the inbound hook.
Called by the OpenClaw agent when a message arrives from the family group.
Checks trigger conditions, routes through the intent engine if triggered,
and returns a result with an optional notification for the group.
Args:
message: The raw text of the inbound Telegram message
sender: Display name of the sender (for notification formatting)
is_reply_to_bot: True if the message is a reply to one of the bot's messages
quoted_text: Text of the message being replied to (if reply-to-bot)
dry_run: If True, parse intent but don't execute mutations
Returns:
A result dict if the message was triggered and processed, or None if
the message didn't meet trigger conditions (should be ignored).
"""
if not check_trigger(message, is_reply_to_bot):
return None
# Strip wake words from the message before intent extraction
clean_message = strip_wake_word(message) if not is_reply_to_bot else message
# Route through the intent router
# If the hook already validated the trigger (wake word found),
# tell the router this is a triggered message by setting is_reply_to_bot=True.
# This bypasses the router's own trigger check since we already validated it.
triggered_by_hook = check_trigger(message, is_reply_to_bot) and not is_reply_to_bot
result = process_group_message(
message=clean_message,
sender=sender,
is_reply_to_bot=is_reply_to_bot or triggered_by_hook,
quoted_message=quoted_text,
dry_run=dry_run,
)
# If the intent engine returned 'none' or 'ignored', this wasn't a
# calendar message after all — don't send a notification
if result.get("status") in ("skipped", "ignored"):
return result
# If we got a notification, it means a calendar action was taken
# The caller (OpenClaw agent) is responsible for sending it to the group
return result
def send_notification_to_group(notification: str) -> bool:
"""Send a notification back to the family Telegram group via Hermes.
Args:
notification: The formatted notification text
Returns:
True if the message was sent successfully
"""
if not notification or not FAMILY_GROUP_ID:
return False
try:
import subprocess
result = subprocess.run(
[
"openclaw", "message", "send",
"--channel", "telegram",
"--target", FAMILY_GROUP_ID,
"--message", notification,
],
capture_output=True,
text=True,
timeout=30,
)
return result.returncode == 0
except Exception as e:
print(f"Error sending notification: {e}")
return False