feat(skills): wire telegram handler.py run() -> bot_operations (FPLAN-0277 live bring-up)

Replace scaffold stub with dispatch table routing all 6 actions (start/stop/
status/create/delete/notify) to bot_operations, bot_factory, notifier. Lazy
imports per action, arg validation, graceful error returns. +28 routing tests
(test_handler_routing.py). Verified live: status -> real registry query.
TG 445/452 (7 telethon-absent), skills 252/252 no regressions. seedgo 25/27
(skill-folder FP + required lazy imports).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-06-15 09:30:58 -07:00
co-authored by Claude Opus 4.8
parent 747c1adf86
commit 5f514604f7
2 changed files with 318 additions and 9 deletions
@@ -1,7 +1,7 @@
# =================== AIPass ====================
# Name: handler.py
# Description: Telegram skill entry point — routes actions to multi-bot framework
# Version: 1.0.0
# Version: 1.1.0
# Created: 2026-06-15
# Modified: 2026-06-15
# =============================================
@@ -12,21 +12,145 @@ telegram — Full 3-layer skill handler.
Routes drone @skills run telegram <action> to the multi-bot framework.
"""
import sys
from pathlib import Path
from aipass.prax import logger
_skill_root = Path(__file__).resolve().parent
if str(_skill_root) not in sys.path:
sys.path.insert(0, str(_skill_root))
_ACTIONS = {
"start",
"stop",
"status",
"create",
"delete",
"notify",
}
def _ok(output: str) -> dict:
return {"success": True, "output": output, "error": None}
def _err(msg: str) -> dict:
return {"success": False, "output": "", "error": msg}
def _cmd_start(args: list) -> dict:
if not args:
return _err("start requires a bot_id: drone @skills run telegram start <bot_id>")
from apps.handlers.bot_operations import start_bot
bot_id = args[0]
exit_code = start_bot(bot_id)
if exit_code is None:
return _err(f"Failed to load config for bot '{bot_id}'")
return _ok(f"Bot '{bot_id}' exited with code {exit_code}")
def _cmd_stop(args: list) -> dict:
if not args:
return _err("stop requires a bot_id: drone @skills run telegram stop <bot_id>")
from apps.handlers.bot_operations import stop_bot
success, message = stop_bot(args[0])
if success:
return _ok(message)
return _err(message)
def _cmd_status(args: list) -> dict:
from apps.handlers.bot_operations import (
format_bot_details,
format_bot_table,
get_status,
)
bot_id = args[0] if args else None
bots = get_status(bot_id)
if not bots:
msg = f"No bot found with id '{bot_id}'" if bot_id else "No bots registered"
return _ok(msg)
if bot_id:
return _ok("\n".join(format_bot_details(bots[0])))
return _ok("\n".join(format_bot_table(bots)))
def _cmd_create(args: list) -> dict:
from apps.handlers.bot_factory import create_bot
from apps.handlers.bot_operations import parse_create_args
parsed = parse_create_args(args)
if not parsed:
return _err("create requires: <bot_id> <token> [--branch name] [--work-dir path]")
result = create_bot(
bot_id=parsed["bot_id"],
bot_token=parsed["bot_token"],
branch_name=parsed.get("branch_name"),
work_dir=parsed.get("work_dir"),
)
if result:
return _ok(f"Bot '{parsed['bot_id']}' created successfully")
return _err(f"Failed to create bot '{parsed['bot_id']}'")
def _cmd_delete(args: list) -> dict:
if not args:
return _err("delete requires a bot_id: drone @skills run telegram delete <bot_id>")
from apps.handlers.bot_factory import delete_bot
success = delete_bot(args[0])
if success:
return _ok(f"Bot '{args[0]}' deleted")
return _err(f"Failed to delete bot '{args[0]}'")
def _cmd_notify(args: list) -> dict:
if not args:
return _err('notify requires a message: drone @skills run telegram notify "message"')
from apps.handlers.notifier import send_telegram_notification
message = " ".join(args)
success = send_telegram_notification(message)
if success:
return _ok(f"Notification sent: {message}")
return _err("Failed to send notification")
_DISPATCH = {
"start": _cmd_start,
"stop": _cmd_stop,
"status": _cmd_status,
"create": _cmd_create,
"delete": _cmd_delete,
"notify": _cmd_notify,
}
def run(action: str, args: list, config: dict) -> dict:
"""
Execute the skill.
Route a skill action to the telegram multi-bot framework.
Args:
action: The action to perform
args: Command arguments
action: The action to perform (start, stop, status, create, delete, notify)
args: Command arguments after the action
config: Skill configuration from SKILL.md
Returns:
dict with keys: success (bool), output (str), error (str|None)
"""
return {
"success": True,
"output": f"telegram executed action: {action}",
"error": None,
}
if not action:
return _err(f"No action specified. Available: {', '.join(sorted(_ACTIONS))}")
handler = _DISPATCH.get(action)
if not handler:
return _err(f"Unknown action '{action}'. Available: {', '.join(sorted(_ACTIONS))}")
try:
return handler(args or [])
except Exception as e:
logger.error("telegram skill action '%s' failed: %s", action, e)
return _err(f"{action} failed: {e}")
@@ -0,0 +1,185 @@
"""Tests for handler.py action routing."""
import sys
from pathlib import Path
from unittest.mock import patch
# Ensure skill root is on path so handler can import apps.handlers.*
_skill_root = Path(__file__).resolve().parents[1]
if str(_skill_root) not in sys.path:
sys.path.insert(0, str(_skill_root))
from handler import run, _ok, _err, _ACTIONS, _DISPATCH
class TestHelpers:
def test_ok_returns_success_dict(self):
result = _ok("hello")
assert result == {"success": True, "output": "hello", "error": None}
def test_err_returns_failure_dict(self):
result = _err("bad thing")
assert result == {"success": False, "output": "", "error": "bad thing"}
class TestRunRouting:
def test_empty_action_returns_error(self):
result = run("", [], {})
assert result["success"] is False
assert "No action specified" in result["error"]
def test_unknown_action_returns_error(self):
result = run("bogus", [], {})
assert result["success"] is False
assert "Unknown action 'bogus'" in result["error"]
assert "create" in result["error"]
def test_all_actions_have_dispatch_entry(self):
for action in _ACTIONS:
assert action in _DISPATCH
def test_dispatch_covers_all_actions(self):
assert set(_DISPATCH.keys()) == _ACTIONS
class TestStartAction:
def test_start_no_args_returns_error(self):
result = run("start", [], {})
assert result["success"] is False
assert "bot_id" in result["error"]
@patch("apps.handlers.bot_operations.start_bot", return_value=0)
def test_start_routes_to_start_bot(self, mock_start):
result = run("start", ["base"], {})
mock_start.assert_called_once_with("base")
assert result["success"] is True
assert "base" in result["output"]
@patch("apps.handlers.bot_operations.start_bot", return_value=None)
def test_start_config_fail_returns_error(self, mock_start):
result = run("start", ["missing"], {})
assert result["success"] is False
assert "missing" in result["error"]
class TestStopAction:
def test_stop_no_args_returns_error(self):
result = run("stop", [], {})
assert result["success"] is False
assert "bot_id" in result["error"]
@patch("apps.handlers.bot_operations.stop_bot", return_value=(True, "Stopped telegram-bot@base"))
def test_stop_success(self, mock_stop):
result = run("stop", ["base"], {})
mock_stop.assert_called_once_with("base")
assert result["success"] is True
@patch("apps.handlers.bot_operations.stop_bot", return_value=(False, "Service not found"))
def test_stop_failure(self, mock_stop):
result = run("stop", ["base"], {})
assert result["success"] is False
class TestStatusAction:
@patch("apps.handlers.bot_operations.get_status", return_value=[])
def test_status_no_bots(self, mock_status):
result = run("status", [], {})
assert result["success"] is True
assert "No bots registered" in result["output"]
@patch("apps.handlers.bot_operations.get_status", return_value=[])
def test_status_specific_bot_not_found(self, mock_status):
result = run("status", ["missing"], {})
assert result["success"] is True
assert "missing" in result["output"]
@patch("apps.handlers.bot_operations.format_bot_details", return_value=["Bot ID: base", "Status: running"])
@patch("apps.handlers.bot_operations.get_status", return_value=[{"bot_id": "base", "status": "running"}])
def test_status_specific_bot_found(self, mock_status, mock_format):
result = run("status", ["base"], {})
assert result["success"] is True
assert "Bot ID: base" in result["output"]
@patch(
"apps.handlers.bot_operations.format_bot_table",
return_value=["Bot ID Branch Status", "base - running"],
)
@patch("apps.handlers.bot_operations.get_status", return_value=[{"bot_id": "base"}, {"bot_id": "dev"}])
def test_status_all_bots(self, mock_status, mock_table):
result = run("status", [], {})
assert result["success"] is True
assert "Bot ID" in result["output"]
class TestCreateAction:
def test_create_no_args_returns_error(self):
result = run("create", [], {})
assert result["success"] is False
assert "requires" in result["error"]
def test_create_missing_token_returns_error(self):
result = run("create", ["mybot"], {})
assert result["success"] is False
@patch("apps.handlers.bot_factory.create_bot", return_value={"bot_id": "mybot"})
def test_create_success(self, mock_create):
result = run("create", ["mybot", "123:ABC"], {})
mock_create.assert_called_once_with(bot_id="mybot", bot_token="123:ABC", branch_name=None, work_dir=None)
assert result["success"] is True
assert "mybot" in result["output"]
@patch("apps.handlers.bot_factory.create_bot", return_value=None)
def test_create_failure(self, mock_create):
result = run("create", ["mybot", "123:ABC"], {})
assert result["success"] is False
@patch("apps.handlers.bot_factory.create_bot", return_value={"bot_id": "mybot"})
def test_create_with_branch_flag(self, mock_create):
result = run("create", ["mybot", "123:ABC", "--branch", "dev"], {})
mock_create.assert_called_once_with(bot_id="mybot", bot_token="123:ABC", branch_name="dev", work_dir=None)
assert result["success"] is True
class TestDeleteAction:
def test_delete_no_args_returns_error(self):
result = run("delete", [], {})
assert result["success"] is False
assert "bot_id" in result["error"]
@patch("apps.handlers.bot_factory.delete_bot", return_value=True)
def test_delete_success(self, mock_delete):
result = run("delete", ["base"], {})
mock_delete.assert_called_once_with("base")
assert result["success"] is True
@patch("apps.handlers.bot_factory.delete_bot", return_value=False)
def test_delete_failure(self, mock_delete):
result = run("delete", ["base"], {})
assert result["success"] is False
class TestNotifyAction:
def test_notify_no_args_returns_error(self):
result = run("notify", [], {})
assert result["success"] is False
assert "message" in result["error"]
@patch("apps.handlers.notifier.send_telegram_notification", return_value=True)
def test_notify_success(self, mock_notify):
result = run("notify", ["hello", "world"], {})
mock_notify.assert_called_once_with("hello world")
assert result["success"] is True
@patch("apps.handlers.notifier.send_telegram_notification", return_value=False)
def test_notify_failure(self, mock_notify):
result = run("notify", ["fail"], {})
assert result["success"] is False
class TestExceptionHandling:
@patch("apps.handlers.bot_operations.get_status", side_effect=RuntimeError("boom"))
def test_exception_caught_and_returned(self, mock_status):
result = run("status", [], {})
assert result["success"] is False
assert "boom" in result["error"]