!/usr/bin/env python3

"""Shadow Mode Runner — Silent observer for Family Logistics group.

This script runs the shadow mode bot that:
1. Observes messages from the Family Logistics Telegram group
2. Runs tripwire detection on every message
3. Extracts entities using LLM when tripwire fires
4. Logs everything to shadow.db
5. NEVER SENDS MESSAGES (hard-disabled)

Usage:
cd /home/hoffmann_admin/.openclaw/workspace/services/icarus
python run_shadow.py

Environment:
Requires shadow.env to be loaded (ENV=shadow)

Safety:
- SPEAK_ENABLED=False is hard-coded
- Database is completely isolated (shadow.db)
- No writes to staging or production
- Bot is invisible to family members
"""

import os
import sys
import asyncio
import logging
from pathlib import Path

Load shadow.env BEFORE anything else reads env vars

This ensures TELEGRAM_BOT_TOKEN is set to the shadow bot token (8442671054)

and NOT inherited from the system env (8705120995 = Wadsworth)

from dotenv import load_dotenv
SCRIPT_DIR = Path(file).parent
env_path = SCRIPT_DIR / "shadow.env"
if env_path.exists():
load_dotenv(env_path, override=True)
print(f"[run_shadow] Loaded env from {env_path}")
else:
print(f"[run_shadow] WARNING: {env_path} not found, using system env")

Add services directory to path for icarus package imports

The icarus package is at /home/hoffmann_admin/.openclaw/workspace/services/icarus

We add services/ so that 'import icarus.core...' works

sys.path.insert(0, '/home/hoffmann_admin/.openclaw/workspace/services')

Set up logging before anything else

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - [SHADOW] - %(levelname)s - %(message)s",
)

Verify shadow environment

env = os.environ.get("ENV", "").lower()
if env != "shadow":
logging.error("=" * 60)
logging.error("CRITICAL: ENV must be 'shadow' to run this script")
logging.error("Current ENV=%r", env)
logging.error("Run with: ENV=shadow python run_shadow.py")
logging.error("=" * 60)
sys.exit(1)

Verify we're not accidentally in wrong directory

if "icarus" not in str(Path.cwd()):
logging.error("CRITICAL: Must run from icarus directory")
sys.exit(1)

Now import shadow modules

from icarus.core.observer import ShadowDatabase, ShadowBot
from icarus.core.observer.shadow_polling import shadow_polling_loop
from icarus.core.config import shadow as shadow_config

Import family context

import yaml

def load_family_context():
"""Load family context from YAML."""
config_path = Path(shadow_config.FAMILY_CONFIG_PATH)
if not config_path.exists():
logging.error("Family config not found: %s", config_path)
return {}

try:
    with open(config_path) as f:
        return yaml.safe_load(f)
except Exception as e:
    logging.error("Failed to load family context: %s", e)
    return {}

async def main():
"""Main shadow mode entry point."""
logging.info("=" * 60)
logging.info("SHADOW MODE INITIALIZING")
logging.info("=" * 60)

# Safety checks
assert not shadow_config.SPEAK_ENABLED, "CRITICAL: SPEAK_ENABLED is True"
assert shadow_config.SHADOW_MODE, "CRITICAL: SHADOW_MODE is False"

logging.info("Safety check passed:")
logging.info("  SPEAK_ENABLED = %s", shadow_config.SPEAK_ENABLED)
logging.info("  SHADOW_MODE = %s", shadow_config.SHADOW_MODE)
logging.info("  DATA_DIR = %s", shadow_config.DATA_DIR)
logging.info("  DB_PATH = %s", shadow_config.SHADOW_DB_PATH)

# Initialize database
logging.info("Initializing shadow database...")
db = ShadowDatabase(shadow_config.SHADOW_DB_PATH)

# Load family context
logging.info("Loading family context...")
family_context = load_family_context()
if family_context:
    members = family_context.get("family", {}).get("members", [])
    logging.info("  Loaded %d family members", len(members))
    for m in members:
        logging.info("    - %s (%s)", m.get("name"), m.get("role"))
else:
    logging.warning("  No family context loaded!")

# Initialize shadow bot
logging.info("Initializing shadow bot...")
verbose_dm = os.environ.get("VERBOSE_ADMIN_DM", "false").lower() == "true"
bot = ShadowBot(
    db=db,
    bot_token=shadow_config.TELEGRAM_BOT_TOKEN,
    family_context=family_context,
    llm_url=shadow_config.LLM_URL,
    llm_model=shadow_config.LLM_MODEL,
    admin_chat_id=shadow_config.ADMIN_CHAT_ID,
    verbose_admin_dm=verbose_dm,
)

logging.info("Shadow bot initialized")
logging.info("  VERBOSE_ADMIN_DM = %s", verbose_dm)
logging.info("Bot will observe messages from TELEGRAM_GROUP_ID=%s", 
             shadow_config.TELEGRAM_GROUP_ID)

# Safety confirmation
logging.info("=" * 60)
logging.info("SHADOW MODE ACTIVE")
logging.info("The bot is now OBSERVING ONLY.")
logging.info("It will NOT send any messages.")
logging.info("It is INVISIBLE to family members.")
logging.info("=" * 60)

# Start polling
try:
    await shadow_polling_loop(bot)
except KeyboardInterrupt:
    logging.info("=" * 60)
    logging.info("SHADOW MODE SHUTTING DOWN")
    logging.info("=" * 60)
except Exception as e:
    logging.exception("Shadow mode error: %s", e)
    raise

if name == "main":
try:
asyncio.run(main())
except Exception as e:
logging.exception("Fatal error: %s", e)
sys.exit(1)