diff --git a/07-event-driven/src/mybot/server/agent_worker.py b/07-event-driven/src/mybot/server/agent_worker.py index 3da4829..fc9fdef 100644 --- a/07-event-driven/src/mybot/server/agent_worker.py +++ b/07-event-driven/src/mybot/server/agent_worker.py @@ -44,13 +44,7 @@ class AgentWorker(SubscriberWorker): except DefNotFoundError as e: logger.error(f"Agent not found: {agent_id}: {e}") - result_event = OutboundEvent( - session_id=event.session_id, - content="", - error=str(e), - ) - await self.context.eventbus.publish(result_event) - return + return await self._emit_response(event, "", str(e)) asyncio.create_task(self.exec_session(event, agent_def)) @@ -76,7 +70,7 @@ class AgentWorker(SubscriberWorker): ) if result: # Emit response and skip agent chat - await self._emit_response(event, result, session, agent_def.id) + await self._emit_response(event, result) logger.info(f"Command completed: {session_id}") return @@ -101,23 +95,20 @@ class AgentWorker(SubscriberWorker): ) await self.context.eventbus.publish(retry_event) else: - result_event = OutboundEvent( - session_id=event.session_id, - content="", - error=str(e), - ) - await self.context.eventbus.publish(result_event) + await self._emit_response(event, "", str(e)) + async def _emit_response( self, event: InboundEvent, content: str, - session, - agent_id: str, + error: str | None = None, ) -> None: """Emit response event with content.""" + result_event = OutboundEvent( session_id=event.session_id, content=content, + error=str(error) if error else None, ) await self.context.eventbus.publish(result_event) diff --git a/08-config-hot-reload/src/mybot/server/agent_worker.py b/08-config-hot-reload/src/mybot/server/agent_worker.py index 3da4829..fc9fdef 100644 --- a/08-config-hot-reload/src/mybot/server/agent_worker.py +++ b/08-config-hot-reload/src/mybot/server/agent_worker.py @@ -44,13 +44,7 @@ class AgentWorker(SubscriberWorker): except DefNotFoundError as e: logger.error(f"Agent not found: {agent_id}: {e}") - result_event = OutboundEvent( - session_id=event.session_id, - content="", - error=str(e), - ) - await self.context.eventbus.publish(result_event) - return + return await self._emit_response(event, "", str(e)) asyncio.create_task(self.exec_session(event, agent_def)) @@ -76,7 +70,7 @@ class AgentWorker(SubscriberWorker): ) if result: # Emit response and skip agent chat - await self._emit_response(event, result, session, agent_def.id) + await self._emit_response(event, result) logger.info(f"Command completed: {session_id}") return @@ -101,23 +95,20 @@ class AgentWorker(SubscriberWorker): ) await self.context.eventbus.publish(retry_event) else: - result_event = OutboundEvent( - session_id=event.session_id, - content="", - error=str(e), - ) - await self.context.eventbus.publish(result_event) + await self._emit_response(event, "", str(e)) + async def _emit_response( self, event: InboundEvent, content: str, - session, - agent_id: str, + error: str | None = None, ) -> None: """Emit response event with content.""" + result_event = OutboundEvent( session_id=event.session_id, content=content, + error=str(error) if error else None, ) await self.context.eventbus.publish(result_event) diff --git a/09-channels/src/mybot/core/context_guard.py b/09-channels/src/mybot/core/context_guard.py index 8739a71..e5bb759 100644 --- a/09-channels/src/mybot/core/context_guard.py +++ b/09-channels/src/mybot/core/context_guard.py @@ -11,6 +11,7 @@ from litellm.types.completion import ( ) from mybot.core.session_state import SessionState +from mybot.utils.config import SourceSessionConfig if TYPE_CHECKING: from mybot.core.context import SharedContext @@ -112,7 +113,7 @@ class ContextGuard: ) -> "SessionState": """Compact history, roll to new session, return new messages.""" new_session = state.agent.new_session(state.source) - self._clear_source_session_cache(str(state.source)) + self._config_source_session_cache(str(state.source), new_session.session_id) compacted_history = await self._build_compacted_messages(state) for message in compacted_history: @@ -155,9 +156,7 @@ class ContextGuard: messages.extend(state.messages[compress_count:]) return messages - def _clear_source_session_cache(self, source_str: str) -> None: - if source_str in self.shared_context.config.sources: - del self.shared_context.config.sources[source_str] - self.shared_context.config.set_runtime( - "sources", self.shared_context.config.sources - ) \ No newline at end of file + def _config_source_session_cache(self, source_str: str, session_id: str) -> None: + self.shared_context.config.set_runtime( + f"""sources.{source_str}""", SourceSessionConfig(session_id=session_id) + ) \ No newline at end of file diff --git a/09-channels/src/mybot/server/agent_worker.py b/09-channels/src/mybot/server/agent_worker.py index 4d11656..308f432 100644 --- a/09-channels/src/mybot/server/agent_worker.py +++ b/09-channels/src/mybot/server/agent_worker.py @@ -45,7 +45,7 @@ class AgentWorker(SubscriberWorker): except DefNotFoundError as e: logger.error(f"Agent not found: {agent_id}: {e}") - await self._emit_response(event, "", agent_def.id, str(e)) + return await self._emit_response(event, "", agent_def.id, str(e)) asyncio.create_task(self.exec_session(event, agent_def)) diff --git a/10-websocket/src/mybot/core/context_guard.py b/10-websocket/src/mybot/core/context_guard.py index 8739a71..e5bb759 100644 --- a/10-websocket/src/mybot/core/context_guard.py +++ b/10-websocket/src/mybot/core/context_guard.py @@ -11,6 +11,7 @@ from litellm.types.completion import ( ) from mybot.core.session_state import SessionState +from mybot.utils.config import SourceSessionConfig if TYPE_CHECKING: from mybot.core.context import SharedContext @@ -112,7 +113,7 @@ class ContextGuard: ) -> "SessionState": """Compact history, roll to new session, return new messages.""" new_session = state.agent.new_session(state.source) - self._clear_source_session_cache(str(state.source)) + self._config_source_session_cache(str(state.source), new_session.session_id) compacted_history = await self._build_compacted_messages(state) for message in compacted_history: @@ -155,9 +156,7 @@ class ContextGuard: messages.extend(state.messages[compress_count:]) return messages - def _clear_source_session_cache(self, source_str: str) -> None: - if source_str in self.shared_context.config.sources: - del self.shared_context.config.sources[source_str] - self.shared_context.config.set_runtime( - "sources", self.shared_context.config.sources - ) \ No newline at end of file + def _config_source_session_cache(self, source_str: str, session_id: str) -> None: + self.shared_context.config.set_runtime( + f"""sources.{source_str}""", SourceSessionConfig(session_id=session_id) + ) \ No newline at end of file diff --git a/10-websocket/src/mybot/server/agent_worker.py b/10-websocket/src/mybot/server/agent_worker.py index 4d11656..308f432 100644 --- a/10-websocket/src/mybot/server/agent_worker.py +++ b/10-websocket/src/mybot/server/agent_worker.py @@ -45,7 +45,7 @@ class AgentWorker(SubscriberWorker): except DefNotFoundError as e: logger.error(f"Agent not found: {agent_id}: {e}") - await self._emit_response(event, "", agent_def.id, str(e)) + return await self._emit_response(event, "", agent_def.id, str(e)) asyncio.create_task(self.exec_session(event, agent_def)) diff --git a/11-multi-agent-routing/src/mybot/cli/main.py b/11-multi-agent-routing/src/mybot/cli/main.py index b5077f7..827e883 100644 --- a/11-multi-agent-routing/src/mybot/cli/main.py +++ b/11-multi-agent-routing/src/mybot/cli/main.py @@ -8,7 +8,6 @@ from rich.console import Console from mybot.cli.chat import chat_command from mybot.cli.server import server_command -from mybot.core.agent_loader import AgentLoader from mybot.utils.config import Config app = typer.Typer( diff --git a/11-multi-agent-routing/src/mybot/core/context_guard.py b/11-multi-agent-routing/src/mybot/core/context_guard.py index 0a45268..6d4111d 100644 --- a/11-multi-agent-routing/src/mybot/core/context_guard.py +++ b/11-multi-agent-routing/src/mybot/core/context_guard.py @@ -112,7 +112,9 @@ class ContextGuard: ) -> "SessionState": """Compact history, roll to new session, return new messages.""" new_session = state.agent.new_session(state.source) - self.shared_context.routing_table.config_source_session_cache(str(state.source), None) + self.shared_context.routing_table.config_source_session_cache( + str(state.source), new_session.session_id + ) compacted_history = await self._build_compacted_messages(state) for message in compacted_history: diff --git a/11-multi-agent-routing/src/mybot/server/agent_worker.py b/11-multi-agent-routing/src/mybot/server/agent_worker.py index 4d11656..308f432 100644 --- a/11-multi-agent-routing/src/mybot/server/agent_worker.py +++ b/11-multi-agent-routing/src/mybot/server/agent_worker.py @@ -45,7 +45,7 @@ class AgentWorker(SubscriberWorker): except DefNotFoundError as e: logger.error(f"Agent not found: {agent_id}: {e}") - await self._emit_response(event, "", agent_def.id, str(e)) + return await self._emit_response(event, "", agent_def.id, str(e)) asyncio.create_task(self.exec_session(event, agent_def)) diff --git a/12-cron-heartbeat/README.md b/12-cron-heartbeat/README.md new file mode 100644 index 0000000..bbc5b28 --- /dev/null +++ b/12-cron-heartbeat/README.md @@ -0,0 +1,205 @@ +# Step 12: Cron + Heartbeat - Scheduled Tasks + +Agent can now initiate conversations on a schedule using CRON.md definitions. + +## Prerequisites + +Same as previous steps - copy the config file and add your API key: + +```bash +cp default_workspace/config.example.yaml default_workspace/config.user.yaml +# Edit config.user.yaml to add your API key +``` + +## What We will Build? + +### Architecture + +``` +CronWorker (every minute) + ↓ +Find due cron jobs + ↓ +Create DispatchEvent + ↓ +AgentWorker executes + ↓ +DispatchResultEvent +``` + +### Key Components + +- **CronLoader**: Loads CRON.md files with schedule definitions +- **CronWorker**: Background worker that checks every minute for due jobs +- **DispatchEvent**: Event type for internal agent-to-agent delegation +- **DispatchResultEvent**: Result event returned from dispatched jobs +- **CronEventSource**: EventSource for cron-triggered events + +## Key Changes + +[src/mybot/core/cron_loader.py](src/mybot/core/cron_loader.py) + +```python +class CronDef(BaseModel): + """Loaded cron job definition.""" + id: str + name: str + description: str + agent: str + schedule: str + prompt: str + one_off: bool = False + + @field_validator("schedule") + @classmethod + def validate_schedule(cls, v: str) -> str: + """Validate cron expression and enforce 5-minute minimum granularity.""" + if not croniter.is_valid(v): + raise ValueError(f"Invalid cron expression: {v}") + # ... validation logic + return v +``` + +[src/mybot/server/cron_worker.py](src/mybot/server/cron_worker.py) + +```python +class CronWorker(Worker): + """Finds due cron jobs, publishes DISPATCH events.""" + + async def run(self) -> None: + """Check every minute for due jobs.""" + self.logger.info("CronWorker started") + + while True: + try: + await self._tick() + except Exception as e: + self.logger.error(f"Error in tick: {e}") + + await asyncio.sleep(60) + + async def _tick(self) -> None: + """Find and dispatch due jobs via EventBus.""" + jobs = self.context.cron_loader.discover_crons() + due_jobs = find_due_jobs(jobs) + + for cron_def in due_jobs: + # Create agent session and dispatch event + event = DispatchEvent( + session_id=session.session_id, + source=CronEventSource(cron_id=cron_def.id), + content=cron_def.prompt, + ) + await self.context.eventbus.publish(event) +``` + +[src/mybot/core/events.py](src/mybot/core/events.py) - New event types + +```python +@dataclass +class CronEventSource(EventSource): + """Source for cron-triggered events.""" + _namespace = "cron" + cron_id: str + +@dataclass +class DispatchEvent(Event): + """Event for internal agent-to-agent delegation.""" + parent_session_id: str = "" + retry_count: int = 0 + +@dataclass +class DispatchResultEvent(Event): + """Event for result of a dispatched job.""" + parent_session_id: str = "" + success: bool = True + error: str | None = None +``` + +[src/mybot/server/agent_worker.py](src/mybot/server/agent_worker.py) - Handle DispatchEvent + +```python +class AgentWorker(SubscriberWorker): + """Dispatches events to session executors.""" + + def __init__(self, context): + super().__init__(context) + # Subscribe to both InboundEvent and DispatchEvent + self.context.eventbus.subscribe(InboundEvent, self.dispatch_event) + self.context.eventbus.subscribe(DispatchEvent, self.dispatch_event) +``` + +## Example CRON.md + +[default_workspace/crons/hello-world/CRON.md](../default_workspace/crons/hello-world/CRON.md) + +```yaml +--- +name: Hello World +description: A simple scheduled task that greets every 5 minutes +agent: default +schedule: "*/5 * * * *" +one_off: false +--- + +Please greet the user and tell them what time it is. This is a scheduled task that runs every 5 minutes. +``` + +**CRON.md Fields:** +- `name`: Human-readable name +- `description`: What this cron job does +- `agent`: Which agent to invoke +- `schedule`: Cron expression (5-field format, minimum 5-minute granularity) +- `one_off`: Delete after first run (default: false) +- Body: Prompt to send to the agent + +## How to Run + +```bash +cd 12-cron-heartbeat +uv run my-bot chat + +# The agent will automatically execute scheduled tasks +# Watch the logs for cron execution messages: +# INFO:mybot.server.cron_worker:Dispatched cron job: hello-world +# INFO:mybot.server.agent_worker:Session completed: ... + +# Create a new cron job: +mkdir -p ../default_workspace/crons/my-task +cat > ../default_workspace/crons/my-task/CRON.md << 'EOF' +--- +name: My Task +description: Runs every 10 minutes +agent: default +schedule: "*/10 * * * *" +--- + +Tell me a fun fact about AI. +EOF + +# The new cron will be picked up automatically on next tick +``` + +## Notes + +**Cron Schedule Format:** +- Uses standard 5-field cron syntax: `minute hour day month weekday` +- Minimum 5-minute granularity enforced (e.g., `*/5 * * * *` works, `*/1 * * * *` fails) +- Examples: + - `*/5 * * * *` - Every 5 minutes + - `0 * * * *` - Every hour + - `0 9 * * 1-5` - Weekdays at 9:00 AM + +**One-off Tasks:** +- Set `one_off: true` to delete the cron after first execution +- Useful for scheduled reminders or delayed tasks + +**Event Flow:** +1. CronWorker checks every minute +2. Creates session and DispatchEvent for due jobs +3. AgentWorker executes the dispatched job +4. DispatchResultEvent emitted (ready for future agent-to-agent communication) + +## What's Next + +[Step 13: Post Message Back](../13-multi-layer-prompts/) - Responsive system prompt diff --git a/12-cron-heartbeat/pyproject.toml b/12-cron-heartbeat/pyproject.toml new file mode 100644 index 0000000..6edee1f --- /dev/null +++ b/12-cron-heartbeat/pyproject.toml @@ -0,0 +1,32 @@ +[project] +name = "my-bot" +version = "0.1.0" +description = "Step 12: Cron + Heartbeat - Scheduled Tasks" +requires-python = ">=3.11" + +dependencies = [ + "litellm>=1.0.0", + "typer>=0.9.0", + "rich>=13.0.0", + "pydantic>=2.0.0", + "pyyaml>=6.0", + "httpx>=0.27.0", + "crawl4ai>=0.3.0", + "watchdog>=3.0.0", + "python-telegram-bot>=20.0", + "discord.py>=2.0", + "fastapi>=0.104.0", + "uvicorn[standard]>=0.24.0", + "websockets>=12.0", + "croniter>=2.0.0", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/mybot"] + +[project.scripts] +my-bot = "mybot.cli.main:app" diff --git a/12-cron-heartbeat/src/mybot/channel/__init__.py b/12-cron-heartbeat/src/mybot/channel/__init__.py new file mode 100644 index 0000000..664342e --- /dev/null +++ b/12-cron-heartbeat/src/mybot/channel/__init__.py @@ -0,0 +1,7 @@ +"""Channel implementations for different platforms.""" + +from mybot.channel.base import Channel +from mybot.channel.telegram_channel import TelegramChannel +from mybot.channel.discord_channel import DiscordChannel + +__all__ = ["Channel", "TelegramChannel", "DiscordChannel"] diff --git a/12-cron-heartbeat/src/mybot/channel/base.py b/12-cron-heartbeat/src/mybot/channel/base.py new file mode 100644 index 0000000..bf4cabe --- /dev/null +++ b/12-cron-heartbeat/src/mybot/channel/base.py @@ -0,0 +1,57 @@ +"""Abstract base class for channel implementations.""" + +from abc import ABC, abstractmethod +from typing import Callable, Awaitable, Generic, TypeVar, Any + +from mybot.core.events import EventSource +from mybot.utils.config import Config + + +T = TypeVar("T", bound=EventSource) + + +class Channel(ABC, Generic[T]): + """Abstract base for messaging platforms with EventSource-based context.""" + + @property + @abstractmethod + def platform_name(self) -> str: + """Platform identifier.""" + pass + + @abstractmethod + async def run(self, on_message: Callable[[str, T], Awaitable[None]]) -> None: + """Run the channel. Blocks until stop() is called.""" + pass + + @abstractmethod + def is_allowed(self, source: T) -> bool: + """Check if sender is whitelisted.""" + pass + + @abstractmethod + async def reply(self, content: str, source: T) -> None: + """Reply to incoming message.""" + pass + + @abstractmethod + async def stop(self) -> None: + """Stop listening and cleanup resources.""" + pass + + @staticmethod + def from_config(config: Config) -> list["Channel[Any]"]: + """Create channel instances from configuration.""" + # Inline imports to avoid circular dependency + from mybot.channel.telegram_channel import TelegramChannel + from mybot.channel.discord_channel import DiscordChannel + + channels: list["Channel[Any]"] = [] + channel_config = config.channels + if channel_config.telegram and channel_config.telegram.enabled: + channels.append(TelegramChannel(channel_config.telegram)) + + if channel_config.discord and channel_config.discord.enabled: + channels.append(DiscordChannel(channel_config.discord)) + + return channels diff --git a/12-cron-heartbeat/src/mybot/channel/discord_channel.py b/12-cron-heartbeat/src/mybot/channel/discord_channel.py new file mode 100644 index 0000000..c2037e5 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/channel/discord_channel.py @@ -0,0 +1,150 @@ +"""Discord channel implementation.""" + +import asyncio +from dataclasses import dataclass +import logging +from typing import Callable, Awaitable + +import discord + +from mybot.core.events import EventSource +from mybot.channel.base import Channel +from mybot.utils.config import DiscordConfig + +logger = logging.getLogger(__name__) + + +@dataclass +class DiscordEventSource(EventSource): + """Source for Discord-originated events.""" + + _namespace = "platform-discord" + user_id: str + channel_id: str + + def __str__(self) -> str: + return f"platform-discord:{self.user_id}:{self.channel_id}" + + @classmethod + def from_string(cls, s: str) -> "DiscordEventSource": + _, user_id, channel_id = s.split(":") + return cls(user_id=user_id, channel_id=channel_id) + + @property + def platform_name(self) -> str: + return "discord" + + +class DiscordChannel(Channel[DiscordEventSource]): + """Discord platform implementation using discord.py.""" + + platform_name = "discord" + + def __init__(self, config: DiscordConfig): + """Initialize DiscordChannel.""" + self.config = config + self.client: discord.Client | None = None + self._running_task: asyncio.Task | None = None + + async def run( + self, on_message: Callable[[str, DiscordEventSource], Awaitable[None]] + ) -> None: + """Run the Discord channel. Blocks until stop() is called.""" + if self._running_task is not None: + raise RuntimeError("DiscordChannel already running") + + logger.info(f"Channel enabled with platform: {self.platform_name}") + + # Configure intents + intents = discord.Intents.default() + intents.message_content = True + intents.messages = True + + self.client = discord.Client(intents=intents) + + @self.client.event + async def _on_discord_message(message: discord.Message) -> None: + """Handle incoming Discord message.""" + # Ignore bot's own messages + if self.client and message.author == self.client.user: + return + + # Check channel restriction (optional) + if ( + self.config.channel_id + and str(message.channel.id) != self.config.channel_id + ): + return + + # Only handle text messages + if not message.content: + return + + # Extract user_id (the person) and channel_id (the channel) + user_id = str(message.author.id) + channel_id = str(message.channel.id) + content = message.content + + logger.info( + f"Received Discord message from user {user_id} in channel {channel_id}" + ) + + source = DiscordEventSource(user_id=user_id, channel_id=channel_id) + + try: + await on_message(content, source) + except Exception as e: + logger.error(f"Error in message callback: {e}") + + # Start the bot and store the task + self._running_task = asyncio.create_task( + self.client.start(self.config.bot_token) + ) + + logger.info("DiscordChannel started") + await self._running_task + + def is_allowed(self, source: DiscordEventSource) -> bool: + """Check if sender is whitelisted.""" + if not self.config.allowed_user_ids: + return True + return source.user_id in self.config.allowed_user_ids + + async def reply(self, content: str, source: DiscordEventSource) -> None: + """Reply to incoming message in the same channel.""" + if not self.client: + raise RuntimeError("DiscordChannel not started") + + try: + channel = self.client.get_channel(int(source.channel_id)) + if not channel: + raise ValueError(f"Channel {source.channel_id} not found") + + # Type ignore: discord.py returns a union, but we know text channels have send() + await channel.send(content) # type: ignore[union-attr] + logger.debug(f"Sent Discord reply to {source.channel_id}") + except Exception as e: + logger.error(f"Failed to send Discord reply: {e}") + raise + + async def stop(self) -> None: + """Stop Discord bot and cleanup.""" + # Idempotent: skip if not running + if self.client is None: + logger.debug("DiscordChannel not running, skipping stop") + return + + await self.client.close() + + # Wait for running task to complete + if self._running_task and not self._running_task.done(): + try: + await asyncio.wait_for(self._running_task, timeout=2.0) + except asyncio.TimeoutError: + logger.warning("Running task did not complete in time") + except Exception: + pass # Task may have already failed + + self.client = None + self._running_task = None + logger.info("DiscordChannel stopped") diff --git a/12-cron-heartbeat/src/mybot/channel/telegram_channel.py b/12-cron-heartbeat/src/mybot/channel/telegram_channel.py new file mode 100644 index 0000000..149957a --- /dev/null +++ b/12-cron-heartbeat/src/mybot/channel/telegram_channel.py @@ -0,0 +1,161 @@ +"""Telegram channel implementation.""" + +import asyncio +from dataclasses import dataclass +import logging +from typing import Callable, Awaitable + +from telegram import Update +from telegram.ext import Application, MessageHandler, filters, ContextTypes + +from mybot.core.events import EventSource +from mybot.channel.base import Channel +from mybot.utils.config import TelegramConfig + +logger = logging.getLogger(__name__) + + +@dataclass +class TelegramEventSource(EventSource): + """Source for Telegram-originated events.""" + + _namespace = "platform-telegram" + user_id: str + chat_id: str + + def __str__(self) -> str: + return f"platform-telegram:{self.user_id}:{self.chat_id}" + + @classmethod + def from_string(cls, s: str) -> "TelegramEventSource": + _, user_id, chat_id = s.split(":") + return cls(user_id=user_id, chat_id=chat_id) + + @property + def platform_name(self) -> str: + return "telegram" + + +class TelegramChannel(Channel[TelegramEventSource]): + """Telegram platform implementation using python-telegram-bot.""" + + platform_name = "telegram" + + def __init__(self, config: TelegramConfig): + """Initialize TelegramChannel.""" + self.config = config + self.application: Application | None = None + self._running_task: asyncio.Task | None = None + self._stop_event: asyncio.Event | None = None + + def is_allowed(self, source: TelegramEventSource) -> bool: + """Check if sender is whitelisted.""" + if not self.config.allowed_user_ids: + return True + return source.user_id in self.config.allowed_user_ids + + async def run( + self, on_message: Callable[[str, TelegramEventSource], Awaitable[None]] + ) -> None: + """Run the Telegram channel. Blocks until stop() is called.""" + if self.application is not None: + raise RuntimeError("TelegramChannel already running") + + logger.info(f"Channel enabled with platform: {self.platform_name}") + self.application = Application.builder().token(self.config.bot_token).build() + self._stop_event = asyncio.Event() + + async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): + """Handle incoming Telegram message.""" + if ( + update.message + and update.message.text + and update.effective_chat + and update.message.from_user + ): + # Extract user_id (the person) and chat_id (the conversation) + user_id = str(update.message.from_user.id) + chat_id = str(update.effective_chat.id) + message = update.message.text + + logger.info( + f"Received Telegram message from user {user_id} in chat {chat_id}" + ) + + source = TelegramEventSource(user_id=user_id, chat_id=chat_id) + + try: + await on_message(message, source) + except Exception as e: + logger.error(f"Error in message callback: {e}") + + handler = MessageHandler(filters.TEXT, handle_message) + self.application.add_handler(handler) + + # Start the bot + await self.application.initialize() + await self.application.start() + if self.application.updater: + await self.application.updater.start_polling() + + logger.info("TelegramChannel started") + + # Create the running task that monitors for stop + async def run_until_stopped(): + """Run until stop() is called or updater stops unexpectedly.""" + while self.application and self.application.updater: + if self.application.updater.running: + if self._stop_event and self._stop_event.is_set(): + return # Graceful stop + await asyncio.sleep(1) + else: + if self._stop_event and not self._stop_event.is_set(): + raise RuntimeError("Telegram updater stopped unexpectedly") + return + + self._running_task = asyncio.create_task(run_until_stopped()) + await self._running_task + + async def reply(self, content: str, source: TelegramEventSource) -> None: + """Reply to incoming message.""" + if not self.application: + raise RuntimeError("TelegramChannel not started") + + try: + await self.application.bot.send_message( + chat_id=int(source.chat_id), text=content + ) + logger.debug(f"Sent Telegram reply to {source.chat_id}") + except Exception as e: + logger.error(f"Failed to send Telegram reply: {e}") + raise + + async def stop(self) -> None: + """Stop Telegram bot and cleanup.""" + # Idempotent: skip if not running + if self.application is None: + logger.debug("TelegramChannel not running, skipping stop") + return + + # Signal the running task to stop + if self._stop_event: + self._stop_event.set() + + if self.application.updater and self.application.updater.running: + await self.application.updater.stop() + await self.application.stop() + await self.application.shutdown() + + # Wait for running task to complete + if self._running_task and not self._running_task.done(): + try: + await asyncio.wait_for(self._running_task, timeout=2.0) + except asyncio.TimeoutError: + logger.warning("Running task did not complete in time") + except Exception: + pass # Task may have already failed + + self.application = None + self._running_task = None + self._stop_event = None + logger.info("TelegramChannel stopped") diff --git a/12-cron-heartbeat/src/mybot/cli/__init__.py b/12-cron-heartbeat/src/mybot/cli/__init__.py new file mode 100644 index 0000000..dc4a0ca --- /dev/null +++ b/12-cron-heartbeat/src/mybot/cli/__init__.py @@ -0,0 +1,5 @@ +"""CLI interface for my-bot.""" + +from mybot.cli.main import app + +__all__ = ["app"] diff --git a/12-cron-heartbeat/src/mybot/cli/chat.py b/12-cron-heartbeat/src/mybot/cli/chat.py new file mode 100644 index 0000000..77dcab9 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/cli/chat.py @@ -0,0 +1,125 @@ +"""Chat CLI command for interactive sessions with event-driven architecture.""" + +import asyncio + +import typer +from rich.console import Console +from rich.panel import Panel +from rich.prompt import Prompt +from rich.text import Text + +from mybot.core.agent import Agent +from mybot.core.context import SharedContext +from mybot.core.events import ( + OutboundEvent, + InboundEvent, + CliEventSource, +) +from mybot.server import ( + AgentWorker, + Worker, +) +from mybot.utils.config import Config, ConfigReloader +from mybot.utils.logging import setup_logging + + +class ChatLoop: + """Interactive chat session using event-driven architecture.""" + + def __init__(self, config: Config, agent_id: str | None = None): + self.config = config + self.console = Console() + self.context = SharedContext(config=config, channels=[]) + self.config_reloader = ConfigReloader(config) + + self.workers: list[Worker] = [ + self.context.eventbus, + AgentWorker(self.context), + ] + + self.response_queue: asyncio.Queue[OutboundEvent] = asyncio.Queue() + self.context.eventbus.subscribe(OutboundEvent, self.handle_outbound_event) + + agent_id = agent_id or config.default_agent + self.agent_def = self.context.agent_loader.load(agent_id) + + async def handle_outbound_event(self, event: OutboundEvent) -> None: + """Handle outbound events by adding to response queue.""" + await self.response_queue.put(event) + self.context.eventbus.ack(event) + + def get_user_input(self) -> str: + """Get user input with styled prompt.""" + prompt_text = Text("You", style="cyan") + user_input = Prompt.ask(prompt_text, console=self.console) + return user_input.strip() + + def display_agent_response(self, content: str) -> None: + """Display agent response with styled prefix.""" + prefix = Text(f"{self.agent_def.id}: ", style="green") + + self.console.print(prefix, end="") + self.console.print(content) + + async def run(self) -> None: + """Run the interactive chat loop.""" + self.console.print( + Panel( + Text("Welcome to my-bot!", style="bold cyan"), + title="Chat", + border_style="cyan", + ) + ) + self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n") + + self.config_reloader.start() + + for worker in self.workers: + worker.start() + + session_id = ( + Agent(self.agent_def, self.context).new_session(CliEventSource()).session_id + ) + + try: + while True: + user_input = await asyncio.to_thread(self.get_user_input) + if user_input.lower() in ("quit", "exit", "q"): + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") + break + + if not user_input: + continue + + event = InboundEvent( + session_id=session_id, + source=CliEventSource(), + content=user_input, + ) + await self.context.eventbus.publish(event) + + try: + response = await asyncio.wait_for( + self.response_queue.get(), timeout=60.0 + ) + + self.display_agent_response(response.content) + except asyncio.TimeoutError: + self.console.print("[red]Agent response timed out[/red]") + self.console.print() + + except (KeyboardInterrupt, EOFError): + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") + finally: + for worker in self.workers: + await worker.stop() + self.config_reloader.stop() + + +def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None: + """Start interactive chat session.""" + config = ctx.obj.get("config") + setup_logging(config, console_output=False) + + chat_loop = ChatLoop(config, agent_id=agent_id) + asyncio.run(chat_loop.run()) diff --git a/12-cron-heartbeat/src/mybot/cli/main.py b/12-cron-heartbeat/src/mybot/cli/main.py new file mode 100644 index 0000000..827e883 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/cli/main.py @@ -0,0 +1,80 @@ +"""CLI interface for my-bot using Typer.""" + +from pathlib import Path +from typing import Annotated + +import typer +from rich.console import Console + +from mybot.cli.chat import chat_command +from mybot.cli.server import server_command +from mybot.utils.config import Config + +app = typer.Typer( + name="my-bot", + help="my-bot: Personal AI Assistant", + no_args_is_help=True, + add_completion=True, +) + +console = Console() + + +def workspace_callback(ctx: typer.Context, workspace: str) -> Path: + """Store workspace path in context for later use.""" + ctx.ensure_object(dict) + ctx.obj["workspace"] = Path(workspace) + return Path(workspace) + + +@app.callback() +def main( + ctx: typer.Context, + workspace: str = typer.Option( + "../default_workspace", + "--workspace", + "-w", + help="Path to workspace directory", + callback=workspace_callback, + ), +) -> None: + """Configuration is loaded from workspace/config.user.yaml by default.""" + workspace_path = ctx.obj["workspace"] + config_file = workspace_path / "config.user.yaml" + + if not config_file.exists(): + console.print(f"[yellow]No configuration found at {config_file}[/yellow]") + raise typer.Exit(1) + + try: + cfg = Config.load(workspace_path) + ctx.obj["config"] = cfg + except Exception as e: + console.print(f"[red]Error loading config: {e}[/red]") + raise typer.Exit(1) + + +@app.command("chat") +def chat( + ctx: typer.Context, + agent: Annotated[ + str | None, + typer.Option( + "--agent", + "-a", + help="Agent ID to use (overrides default_agent from config)", + ), + ] = None, +) -> None: + """Start interactive chat session.""" + chat_command(ctx, agent_id=agent) + + +@app.command("server") +def server(ctx: typer.Context) -> None: + """Start the 24/7 server for cron and messagebus execution.""" + server_command(ctx) + + +if __name__ == "__main__": + app() diff --git a/12-cron-heartbeat/src/mybot/cli/server.py b/12-cron-heartbeat/src/mybot/cli/server.py new file mode 100644 index 0000000..8b5369f --- /dev/null +++ b/12-cron-heartbeat/src/mybot/cli/server.py @@ -0,0 +1,25 @@ +"""Server CLI command for worker-based architecture.""" + +import asyncio + +import typer + +from mybot.core.context import SharedContext +from mybot.server.server import Server +from mybot.utils.logging import setup_logging + + +def server_command(ctx: typer.Context) -> None: + """Start the 24/7 server for cron and messagebus execution.""" + config = ctx.obj.get("config") + + setup_logging(config, console_output=True) + + typer.echo("Starting mybot server...") + typer.echo("Press Ctrl+C to stop") + + try: + context = SharedContext(config) + asyncio.run(Server(context).run()) + except KeyboardInterrupt: + typer.echo("\nServer stopped") diff --git a/12-cron-heartbeat/src/mybot/core/__init__.py b/12-cron-heartbeat/src/mybot/core/__init__.py new file mode 100644 index 0000000..762fef9 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/__init__.py @@ -0,0 +1,23 @@ +"""Core agent functionality.""" + +from .agent import Agent, AgentSession +from .agent_loader import ( + AgentLoader, + AgentDef, +) +from .context import SharedContext +from .history import HistoryMessage, HistorySession, HistoryStore +from .routing import Binding, RoutingTable + +__all__ = [ + "Agent", + "AgentSession", + "AgentDef", + "AgentLoader", + "SharedContext", + "HistoryStore", + "HistoryMessage", + "HistorySession", + "Binding", + "RoutingTable", +] diff --git a/12-cron-heartbeat/src/mybot/core/agent.py b/12-cron-heartbeat/src/mybot/core/agent.py new file mode 100644 index 0000000..9c73aba --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/agent.py @@ -0,0 +1,235 @@ +import uuid +import json +import asyncio +from dataclasses import dataclass, field +from datetime import datetime +from typing import TYPE_CHECKING + +from mybot.core.context_guard import ContextGuard +from mybot.core.session_state import SessionState +from mybot.core.events import EventSource +from mybot.provider.llm import LLMProvider +from mybot.tools.registry import ToolRegistry +from mybot.tools.skill_tool import create_skill_tool +from mybot.tools.websearch_tool import create_websearch_tool +from mybot.tools.webread_tool import create_webread_tool + +from litellm.types.completion import ( + ChatCompletionMessageParam as Message, + ChatCompletionMessageToolCallParam, +) + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + from mybot.core.agent_loader import AgentDef + from mybot.provider.llm import LLMToolCall + + +class Agent: + """A configured agent that creates and manages conversation sessions.""" + + def __init__(self, agent_def: "AgentDef", context: "SharedContext") -> None: + self.agent_def = agent_def + self.context = context + self.llm = LLMProvider.from_config(agent_def.llm) + + def _build_tools(self) -> ToolRegistry: + """Build a ToolRegistry with tools appropriate for the session.""" + registry = ToolRegistry.with_builtins() + + # Register skill tool if allowed + if self.agent_def.allow_skills: + skill_tool = create_skill_tool(self.context.skill_loader) + if skill_tool: + registry.register(skill_tool) + + websearch_tool = create_websearch_tool(self.context) + if websearch_tool: + registry.register(websearch_tool) + + webread_tool = create_webread_tool(self.context) + if webread_tool: + registry.register(webread_tool) + + return registry + + def _get_token_threshold(self) -> int: + """Get token threshold based on model's context window.""" + # Default to 80% of 200k context + return 160000 + + def new_session( + self, + source: EventSource, + session_id: str | None = None, + ) -> "AgentSession": + """Create a new conversation session.""" + session_id = session_id or str(uuid.uuid4()) + tools = self._build_tools() + + # Create context guard for this session + context_guard = ContextGuard( + shared_context=self.context, + token_threshold=self._get_token_threshold(), + ) + + state = SessionState( + session_id=session_id, + agent=self, + messages=[], + source=source, + shared_context=self.context, + ) + + session = AgentSession( + agent=self, + state=state, + context_guard=context_guard, + tools=tools, + ) + + self.context.history_store.create_session( + self.agent_def.id, session_id, source + ) + return session + + def resume_session(self, session_id: str) -> "AgentSession": + """Load an existing conversation session.""" + session_query = [ + session + for session in self.context.history_store.list_sessions() + if session.id == session_id + ] + if not session_query: + raise ValueError(f"Session not found: {session_id}") + + session_info = session_query[0] + source = session_info.get_source() + + # Get all messages (no max_history limit) + history_messages = self.context.history_store.get_messages(session_id) + + # Convert HistoryMessage to litellm Message format + messages: list[Message] = [msg.to_message() for msg in history_messages] + + # Build tools for resumed session + tools = self._build_tools() + + # Create context guard + context_guard = ContextGuard( + shared_context=self.context, + token_threshold=self._get_token_threshold(), + ) + + # Create SessionState with loaded messages + state = SessionState( + session_id=session_info.id, + agent=self, + messages=messages, + source=source, + shared_context=self.context, + ) + + return AgentSession( + agent=self, + state=state, + context_guard=context_guard, + tools=tools, + ) + + +@dataclass +class AgentSession: + """Chat orchestrator - operates on swappable SessionState.""" + + agent: Agent + state: SessionState + context_guard: ContextGuard + tools: ToolRegistry + started_at: datetime = field(default_factory=datetime.now) + + @property + def session_id(self) -> str: + """Delegate to state.""" + return self.state.session_id + + @property + def source(self) -> "EventSource": + return self.state.source + + @property + def shared_context(self) -> "SharedContext": + """Delegate to state.""" + return self.state.shared_context + + async def chat(self, message: str) -> str: + """Send a message to the LLM and get a response.""" + user_msg: Message = {"role": "user", "content": message} + self.state.add_message(user_msg) + + tool_schemas = self.tools.get_tool_schemas() + + while True: + messages = self.state.build_messages() + self.state = await self.context_guard.check_and_compact(self.state) + content, tool_calls = await self.agent.llm.chat(messages, tool_schemas) + + tool_call_dicts: list[ChatCompletionMessageToolCallParam] = [ + { + "id": tc.id, + "type": "function", + "function": {"name": tc.name, "arguments": tc.arguments}, + } + for tc in tool_calls + ] + assistant_msg: Message = { + "role": "assistant", + "content": content, + "tool_calls": tool_call_dicts, + } + + self.state.add_message(assistant_msg) + + if not tool_calls: + break + + await self._handle_tool_calls(tool_calls) + + continue + + return content + + async def _handle_tool_calls( + self, + tool_calls: list["LLMToolCall"], + ) -> None: + """Handle tool calls from the LLM response.""" + tool_call_results = await asyncio.gather( + *[self._execute_tool_call(tool_call) for tool_call in tool_calls] + ) + + for tool_call, result in zip(tool_calls, tool_call_results): + tool_msg: Message = { + "role": "tool", + "content": result, + "tool_call_id": tool_call.id, + } + self.state.add_message(tool_msg) + + async def _execute_tool_call( + self, + tool_call: "LLMToolCall", + ) -> str: + """Execute a single tool call.""" + # Extract key arguments + try: + args = json.loads(tool_call.arguments) + except json.JSONDecodeError: + args = {} + + try: + result = await self.tools.execute_tool(tool_call.name, session=self, **args) + except Exception as e: + result = f"Error executing tool: {e}" + + return result diff --git a/12-cron-heartbeat/src/mybot/core/agent_loader.py b/12-cron-heartbeat/src/mybot/core/agent_loader.py new file mode 100644 index 0000000..593f63c --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/agent_loader.py @@ -0,0 +1,84 @@ +"""Agent definition loader.""" + +from typing import Any + +from pydantic import BaseModel, ValidationError + +from mybot.utils.config import Config, LLMConfig +from mybot.utils.def_loader import ( + DefNotFoundError, + InvalidDefError, + discover_definitions, + parse_definition, +) + + +class AgentDef(BaseModel): + """Loaded agent definition with merged settings.""" + + id: str + name: str + description: str = "" + agent_md: str + llm: LLMConfig + allow_skills: bool = False + + +class AgentLoader: + """Loads agent definitions from AGENT.md files.""" + + @staticmethod + def from_config(config: Config) -> "AgentLoader": + return AgentLoader(config) + + def __init__(self, config: Config): + """Initialize AgentLoader.""" + self.config = config + + def load(self, agent_id: str) -> AgentDef: + """Load agent by ID.""" + agent_file = self.config.agents_path / agent_id / "AGENT.md" + if not agent_file.exists(): + raise DefNotFoundError("agent", agent_id) + + try: + content = agent_file.read_text() + agent_def = parse_definition(content, agent_id, self._parse_agent_def) + except InvalidDefError: + raise + except Exception as e: + raise InvalidDefError("agent", agent_id, str(e)) + + return agent_def + + def discover_agents(self) -> list[AgentDef]: + """Scan agents directory and return list of valid AgentDef.""" + return discover_definitions( + self.config.agents_path, "AGENT.md", self._parse_agent_def + ) + + def _parse_agent_def( + self, def_id: str, frontmatter: dict[str, Any], body: str + ) -> AgentDef: + """Parse agent definition from frontmatter (callback for parse_definition).""" + llm_overrides = frontmatter.get("llm") + merged_llm = self._merge_llm_config(llm_overrides) + + try: + return AgentDef( + id=def_id, + name=frontmatter["name"], # type: ignore[misc] + description=frontmatter.get("description", ""), + agent_md=body.strip(), + llm=merged_llm, + allow_skills=frontmatter.get("allow_skills", False), + ) + except ValidationError as e: + raise InvalidDefError("agent", def_id, str(e)) + + def _merge_llm_config(self, agent_llm: dict[str, Any] | None) -> LLMConfig: + """Deep merge agent's llm config with global defaults.""" + base = self.config.llm.model_dump() + if agent_llm: + base = {**base, **agent_llm} + return LLMConfig(**base) diff --git a/12-cron-heartbeat/src/mybot/core/commands/__init__.py b/12-cron-heartbeat/src/mybot/core/commands/__init__.py new file mode 100644 index 0000000..71b0ea7 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/commands/__init__.py @@ -0,0 +1,6 @@ +"""Slash commands module.""" + +from mybot.core.commands.base import Command +from mybot.core.commands.registry import CommandRegistry + +__all__ = ["Command", "CommandRegistry"] diff --git a/12-cron-heartbeat/src/mybot/core/commands/base.py b/12-cron-heartbeat/src/mybot/core/commands/base.py new file mode 100644 index 0000000..588f3d2 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/commands/base.py @@ -0,0 +1,20 @@ +"""Base classes for slash commands.""" + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class Command(ABC): + """Base class for slash commands.""" + + name: str + aliases: list[str] = [] + description: str = "" + + @abstractmethod + async def execute(self, args: str, session: "AgentSession") -> str: + """Execute the command and return response string.""" + pass diff --git a/12-cron-heartbeat/src/mybot/core/commands/handlers.py b/12-cron-heartbeat/src/mybot/core/commands/handlers.py new file mode 100644 index 0000000..f2b57cf --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/commands/handlers.py @@ -0,0 +1,248 @@ +"""Built-in slash command handlers.""" + +from typing import TYPE_CHECKING + +from mybot.core.commands.base import Command +from mybot.utils.def_loader import DefNotFoundError + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class SessionCommand(Command): + """Show current session details.""" + + name = "session" + description = "Show current session details" + + async def execute(self, args: str, session: "AgentSession") -> str: + info = session.shared_context.history_store.get_session_info(session.session_id) + + # Handle case where session not found in indexs + created_str = info.created_at if info else "Unknown" + + lines = [ + f"**Session ID:** `{session.session_id}`", + f"**Agent:** {session.agent.agent_def.name} (`{session.agent.agent_def.id}`)", + f"**Created:** {created_str}", + f"**Messages:** {len(session.state.messages)}", + f"**Source:** `{session.source}`", + ] + return "\n".join(lines) + + +class HelpCommand(Command): + """Show available commands.""" + + name = "help" + aliases = ["?"] + description = "Show available commands" + + async def execute(self, args: str, session: "AgentSession") -> str: + lines = ["**Available Commands:**"] + for cmd in session.shared_context.command_registry.list_commands(): + names = [f"/{cmd.name}"] + [f"/{a}" for a in cmd.aliases] + lines.append(f"{', '.join(names)} - {cmd.description}") + return "\n".join(lines) + + +class CompactCommand(Command): + """Trigger manual context compaction.""" + + name = "compact" + description = "Compact conversation context manually" + + async def execute(self, args: str, session: "AgentSession") -> str: + # Force compaction regardless of threshold + await session.context_guard._compact_messages(session.state) + msg_count = len(session.state.messages) + return f"✓ Context compacted. {msg_count} messages retained." + + +class ContextCommand(Command): + """Show session context information.""" + + name = "context" + description = "Show session context information" + + async def execute(self, args: str, session: "AgentSession") -> str: + token_count = session.context_guard.estimate_tokens(session.state) + threshold = session.context_guard.token_threshold + usage_pct = (token_count / threshold) * 100 if threshold > 0 else 0 + + lines = [ + f"**Messages:** {len(session.state.messages)}", + f"**Tokens:** {token_count:,} ({usage_pct:.1f}% of {threshold:,} threshold)", + ] + return "\n".join(lines) + + +class ClearCommand(Command): + """Clear conversation and start fresh.""" + + name = "clear" + description = "Clear conversation and start fresh" + + async def execute(self, args: str, session: "AgentSession") -> str: + source_str = str(session.source) + + session.shared_context.routing_table.config_source_session_cache(source_str, None) + + return "✓ Conversation cleared. Next message starts fresh." + + +class AgentCommand(Command): + """List agents or show agent details.""" + + name = "agent" + aliases = ["agents"] + description = "List agents or show agent details" + + async def execute(self, args: str, session: "AgentSession") -> str: + if not args: + # List agents + agents = session.shared_context.agent_loader.discover_agents() + lines = ["**Agents:**"] + for agent in agents: + marker = " (current)" if agent.id == session.agent.agent_def.id else "" + lines.append(f"- `{agent.id}`: {agent.name}{marker}") + return "\n".join(lines) + + # Show specific agent details + agent_id = args.strip() + try: + agent_def = session.shared_context.agent_loader.load(agent_id) + except ValueError: + return f"✗ Agent `{agent_id}` not found." + + lines = [ + f"**Agent:** `{agent_def.id}`", + f"**Name:** {agent_def.name}", + f"**Description:** {agent_def.description}", + f"**LLM:** {agent_def.llm.model}", + ] + + # Add content sections + lines.append(f"\n---\n\n**AGENT.md:**\n```\n{agent_def.agent_md}\n```") + + if agent_def.soul_md: + lines.append(f"\n**SOUL.md:**\n```\n{agent_def.soul_md}\n```") + + return "\n".join(lines) + + +class SkillsCommand(Command): + """List all skills or show skill details.""" + + name = "skills" + description = "List all skills or show skill details" + + async def execute(self, args: str, session: "AgentSession") -> str: + if not args: + skills = session.shared_context.skill_loader.discover_skills() + if not skills: + return "No skills configured." + + lines = ["**Skills:**"] + for skill in skills: + lines.append(f"- `{skill.id}`: {skill.description}") + return "\n".join(lines) + + # Show specific skill details + skill_id = args.strip() + try: + skill = session.shared_context.skill_loader.load_skill(skill_id) + except DefNotFoundError: + return f"✗ Skill `{skill_id}` not found." + + lines = [ + f"**Skill:** `{skill.id}`", + f"**Name:** {skill.name}", + f"**Description:** {skill.description}", + f"\n---\n\n**SKILL.md:**\n```\n{skill.content}\n```", + ] + return "\n".join(lines) + + +class CronsCommand(Command): + """List all cron jobs or show cron details.""" + + name = "crons" + description = "List all cron jobs or show cron details" + + async def execute(self, args: str, session: "AgentSession") -> str: + if not args: + crons = session.shared_context.cron_loader.discover_crons() + if not crons: + return "No cron jobs configured." + + lines = ["**Cron Jobs:**"] + for cron in crons: + lines.append(f"- `{cron.id}`: {cron.schedule}") + return "\n".join(lines) + + # Show specific cron details + cron_id = args.strip() + try: + cron = session.shared_context.cron_loader.load(cron_id) + except DefNotFoundError: + return f"✗ Cron `{cron_id}` not found." + + lines = [ + f"**Cron:** `{cron.id}`", + f"**Name:** {cron.name}", + f"**Schedule:** `{cron.schedule}`", + f"**Agent:** {cron.agent}", + f"\n---\n\n**CRON.md:**\n```\n{cron.prompt}\n```", + ] + return "\n".join(lines) + + +class RouteCommand(Command): + """Create a routing binding.""" + + name = "route" + description = "Create a routing binding (persists to config)" + + async def execute(self, args: str, session: "AgentSession") -> str: + parts = args.strip().split(None, 1) + if len(parts) != 2: + return "**Usage:** `/route `\n\nExample: `/route platform-telegram:.* pickle`" + + pattern, agent_id = parts + + # Validate regex pattern + try: + re.compile(f"^{pattern}$") + except re.error as e: + return f"✗ Invalid regex pattern: {e}" + + # Verify agent exists + try: + session.shared_context.agent_loader.load(agent_id) + except ValueError: + return f"✗ Agent `{agent_id}` not found." + + # Create and persist binding + session.shared_context.routing_table.persist_binding(pattern, agent_id) + + return f"✓ Route bound: `{pattern}` → `{agent_id}`" + + +class BindingsCommand(Command): + """Show all routing bindings.""" + + name = "bindings" + description = "Show all routing bindings" + + async def execute(self, args: str, session: "AgentSession") -> str: + bindings = session.shared_context.config.routing.get("bindings", []) + + if not bindings: + return "No routing bindings configured." + + lines = ["**Routing Bindings:**"] + for binding in bindings: + lines.append(f"- `{binding['value']}` → `{binding['agent']}`") + + return "\n".join(lines) diff --git a/12-cron-heartbeat/src/mybot/core/commands/registry.py b/12-cron-heartbeat/src/mybot/core/commands/registry.py new file mode 100644 index 0000000..57a72bc --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/commands/registry.py @@ -0,0 +1,86 @@ +"""Command registry for managing slash commands.""" + +from typing import TYPE_CHECKING + +from mybot.core.commands.base import Command + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class CommandRegistry: + """Registry for slash commands.""" + + def __init__(self) -> None: + self._commands: dict[str, Command] = {} + + def register(self, cmd: Command) -> None: + """Register a command and its aliases.""" + self._commands[cmd.name] = cmd + for alias in cmd.aliases: + self._commands[alias] = cmd + + def list_commands(self) -> list[Command]: + """Return list of unique commands (deduplicated by name).""" + seen = set() + commands = [] + for cmd in self._commands.values(): + if cmd.name not in seen: + seen.add(cmd.name) + commands.append(cmd) + return commands + + def resolve(self, input: str) -> tuple[Command, str] | None: + """Parse input and return (command, args) if it matches.""" + if not input.startswith("/"): + return None + + parts = input[1:].split(None, 1) + if not parts: + return None + + cmd_name = parts[0].lower() + args = parts[1] if len(parts) > 1 else "" + + cmd = self._commands.get(cmd_name) + if cmd: + return (cmd, args) + return None + + async def dispatch(self, input: str, session: "AgentSession") -> str | None: + """Parse and execute a slash command.""" + resolved = self.resolve(input) + if not resolved: + return None + + cmd, args = resolved + return await cmd.execute(args, session) + + @classmethod + def with_builtins(cls) -> "CommandRegistry": + """Create registry with built-in commands registered.""" + from mybot.core.commands.handlers import ( + HelpCommand, + AgentCommand, + SkillsCommand, + CronsCommand, + CompactCommand, + ContextCommand, + ClearCommand, + SessionCommand, + RouteCommand, + BindingsCommand, + ) + + registry = cls() + registry.register(HelpCommand()) + registry.register(AgentCommand()) + registry.register(SkillsCommand()) + registry.register(CronsCommand()) + registry.register(CompactCommand()) + registry.register(ContextCommand()) + registry.register(ClearCommand()) + registry.register(SessionCommand()) + registry.register(RouteCommand()) + registry.register(BindingsCommand()) + return registry diff --git a/12-cron-heartbeat/src/mybot/core/context.py b/12-cron-heartbeat/src/mybot/core/context.py new file mode 100644 index 0000000..1365134 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/context.py @@ -0,0 +1,48 @@ +from typing import Any, TYPE_CHECKING + +from mybot.core.agent_loader import AgentLoader +from mybot.core.commands.registry import CommandRegistry +from mybot.core.cron_loader import CronLoader +from mybot.core.history import HistoryStore +from mybot.core.routing import RoutingTable +from mybot.core.skill_loader import SkillLoader +from mybot.core.eventbus import EventBus +from mybot.channel.base import Channel +from mybot.utils.config import Config + +if TYPE_CHECKING: + from mybot.server.websocket_worker import WebSocketWorker + + +class SharedContext: + """Global shared state for the application.""" + + config: Config + history_store: HistoryStore + agent_loader: AgentLoader + skill_loader: SkillLoader + cron_loader: CronLoader + command_registry: CommandRegistry + routing_table: RoutingTable + channels: list[Channel[Any]] + eventbus: EventBus + websocket_worker: "WebSocketWorker | None" + + def __init__( + self, config: Config, channels: list[Channel[Any]] | None = None + ) -> None: + self.config = config + self.history_store = HistoryStore.from_config(config) + self.agent_loader = AgentLoader.from_config(config) + self.skill_loader = SkillLoader.from_config(config) + self.cron_loader = CronLoader.from_config(config) + self.command_registry = CommandRegistry.with_builtins() + self.routing_table = RoutingTable(self) + + if channels is not None: + self.channels = channels + else: + self.channels = Channel.from_config(config) + + self.eventbus = EventBus(self) + self.websocket_worker = None diff --git a/12-cron-heartbeat/src/mybot/core/context_guard.py b/12-cron-heartbeat/src/mybot/core/context_guard.py new file mode 100644 index 0000000..e4c570c --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/context_guard.py @@ -0,0 +1,158 @@ +"""Context guard for proactive context window management.""" + +from dataclasses import dataclass +from typing import TYPE_CHECKING, cast + +from litellm import token_counter +from litellm.types.completion import ( + ChatCompletionMessageParam as Message, + ChatCompletionAssistantMessageParam, + ChatCompletionToolMessageParam, +) + +from mybot.core.session_state import SessionState + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + from mybot.core.session_state import SessionState + + +# Default max size for tool result content before truncation +MAX_TOOL_RESULT_CHARS = 10000 + + +@dataclass +class ContextGuard: + """Manages context window size with proactive compaction.""" + + shared_context: "SharedContext" + token_threshold: int = 160000 # 80% of 200k context + max_tool_result_chars: int = MAX_TOOL_RESULT_CHARS + + def estimate_tokens(self, state: "SessionState") -> int: + """Estimate token count for session state.""" + if not state.messages: + return 0 + return token_counter( + model=state.agent.agent_def.llm.model, messages=state.build_messages() + ) + + async def check_and_compact( + self, + state: "SessionState", + ) -> "SessionState": + """Check token count, compact and roll session if needed.""" + token_count = self.estimate_tokens(state) + + if token_count < self.token_threshold: + return state + + state.messages = self._truncate_large_tool_results(state.messages) + token_count = self.estimate_tokens(state) + + if token_count < self.token_threshold: + return state + + return await self.compact_and_roll(state) + + def _compress_message_count(self, state: "SessionState") -> int: + keep_count = max(4, int(len(state.messages) * 0.2)) + compress_count = max(2, int(len(state.messages) * 0.5)) + return min(compress_count, len(state.messages) - keep_count) + + def _truncate_large_tool_results(self, messages: list[Message]) -> list[Message]: + """Truncate oversized tool results to reduce context size.""" + result: list[Message] = [] + for msg in messages: + if msg.get("role") == "tool": + content = msg.get("content", "") + if ( + isinstance(content, str) + and len(content) > self.max_tool_result_chars + ): + original_size = len(content) + truncated = content[: self.max_tool_result_chars] + truncated_content = ( + f"{truncated}\n\n" + f"[Truncated - original size: {original_size} chars]" + ) + + msg = cast( + ChatCompletionToolMessageParam, + {**msg, "content": truncated_content}, + ) + + result.append(msg) + return result + + def _serialize_messages_for_summary(self, messages: list[Message]) -> str: + """Serialize messages to plain text for summarization.""" + lines = [] + for msg in messages: + role = msg.get("role", "unknown") + content = msg.get("content", "") + # Handle tool calls in assistant messages + if role == "assistant" and msg.get("tool_calls"): + tool_names = [ + tc.get("function", {}).get("name", "unknown") + for tc in (cast(ChatCompletionAssistantMessageParam, msg)).get( + "tool_calls", [] + ) + ] + lines.append( + f"ASSISTANT: [used tools: {', '.join(tool_names)}] {content}" + ) + else: + lines.append(f"{role.upper()}: {content}") + return "\n".join(lines) + + async def compact_and_roll( + self, + state: "SessionState", + ) -> "SessionState": + """Compact history, roll to new session, return new messages.""" + new_session = state.agent.new_session(state.source) + self.shared_context.routing_table.config_source_session_cache( + str(state.source), new_session.session_id + ) + + compacted_history = await self._build_compacted_messages(state) + for message in compacted_history: + new_session.state.add_message(message) + + return new_session.state + + async def _build_compacted_messages( + self, + state: "SessionState", + ) -> list[Message]: + """Generate summary of older messages using agent's LLM.""" + compress_count = self._compress_message_count(state) + + old_messages = state.messages[:compress_count] + old_text = self._serialize_messages_for_summary(old_messages) + + summary_prompt = f"""Summarize the conversation so far. Keep it factual and concise. Focus on key decisions, facts, and user preferences discovered: + +{old_text}""" + + response, _ = await state.agent.llm.chat( + [{"role": "user", "content": summary_prompt}], + [], # No tools needed + ) + + messages: list[Message] = [] + messages.append( + { + "role": "user", + "content": f"[Previous conversation summary]\n{response}", + } + ) + messages.append( + { + "role": "assistant", + "content": "Understood, I have the context.", + } + ) + messages.extend(state.messages[compress_count:]) + return messages diff --git a/12-cron-heartbeat/src/mybot/core/cron_loader.py b/12-cron-heartbeat/src/mybot/core/cron_loader.py new file mode 100644 index 0000000..90a66c7 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/cron_loader.py @@ -0,0 +1,113 @@ +"""Cron job definition loader.""" + +import logging +from datetime import datetime +from typing import TYPE_CHECKING, Any + +from croniter import croniter +from pydantic import BaseModel, ValidationError, field_validator + +from mybot.utils.def_loader import ( + DefNotFoundError, + InvalidDefError, + discover_definitions, + parse_definition, +) + +if TYPE_CHECKING: + from mybot.utils.config import Config + +logger = logging.getLogger(__name__) + + +class CronDef(BaseModel): + """Loaded cron job definition.""" + + id: str + name: str + description: str + agent: str + schedule: str + prompt: str + one_off: bool = False + + @field_validator("schedule") + @classmethod + def validate_schedule(cls, v: str) -> str: + """Validate cron expression and enforce 5-minute minimum granularity.""" + if not croniter.is_valid(v): + raise ValueError(f"Invalid cron expression: {v}") + + # Check minimum 5-minute granularity using croniter + # Get the first two run times and check the gap + base = datetime(2024, 1, 1, 0, 0) # Arbitrary base time + cron = croniter(v, base) + first_run = cron.get_next(datetime) + second_run = cron.get_next(datetime) + gap_minutes = (second_run - first_run).total_seconds() / 60 + + if gap_minutes < 5: + raise ValueError( + f"Schedule must have minimum 5-minute granularity. Got: {v} (runs every {gap_minutes:.0f} min)" + ) + + return v + + +class CronLoader: + """Loads cron job definitions from CRON.md files.""" + + @staticmethod + def from_config(config: "Config") -> "CronLoader": + """Create CronLoader from config.""" + return CronLoader(config) + + def __init__(self, config: "Config"): + """Initialize CronLoader.""" + self.config = config + + def discover_crons(self) -> list[CronDef]: + """Scan crons directory, return definitions for all valid jobs.""" + return discover_definitions( + self.config.crons_path, "CRON.md", self._parse_cron_def + ) + + def _parse_cron_def( + self, def_id: str, frontmatter: dict[str, Any], body: str + ) -> CronDef | None: + """Parse cron definition from frontmatter (callback for discover_definitions).""" + try: + return CronDef( + id=def_id, + name=frontmatter["name"], # type: ignore[misc] + description=frontmatter["description"], # type: ignore[misc] + agent=frontmatter["agent"], # type: ignore[misc] + schedule=frontmatter["schedule"], # type: ignore[misc] + prompt=body.strip(), + one_off=frontmatter.get("one_off", False), + ) + except ValidationError as e: + logger.warning(f"Invalid cron '{def_id}': {e}") + return None + except KeyError as e: + logger.warning(f"Missing required field in cron '{def_id}': {e}") + return None + + def load(self, cron_id: str) -> CronDef: + """Load cron by ID.""" + cron_file = self.config.crons_path / cron_id / "CRON.md" + if not cron_file.exists(): + raise DefNotFoundError("cron", cron_id) + + try: + content = cron_file.read_text() + cron_def = parse_definition(content, cron_id, self._parse_cron_def) + except InvalidDefError: + raise + except Exception as e: + raise InvalidDefError("cron", cron_id, str(e)) + + if cron_def is None: + raise InvalidDefError("cron", cron_id, "validation failed") + + return cron_def diff --git a/12-cron-heartbeat/src/mybot/core/eventbus.py b/12-cron-heartbeat/src/mybot/core/eventbus.py new file mode 100644 index 0000000..af3227e --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/eventbus.py @@ -0,0 +1,145 @@ +"""Central event bus for pub/sub event distribution.""" + +import asyncio +import json +import logging +import os +from collections import defaultdict +from typing import Awaitable, Callable, TypeVar, TYPE_CHECKING + +from mybot.server.worker import Worker + +from .events import ( + Event, + OutboundEvent, + deserialize_event, +) + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + +logger = logging.getLogger(__name__) + +E = TypeVar("E", bound=Event) +Handler = Callable[[Event], Awaitable[None]] + + +class EventBus(Worker): + """Central event bus with subscription support and async dispatch.""" + + def __init__(self, context: "SharedContext"): + super().__init__(context) + self.context = context + self._subscribers: dict[type[Event], list[Handler]] = defaultdict(list) + self._queue: asyncio.Queue[Event] = asyncio.Queue() + self.pending_dir = context.config.event_path / "pending" + self.pending_dir.mkdir(parents=True, exist_ok=True) + + def subscribe( + self, event_class: type[E], handler: Callable[[E], Awaitable[None]] + ) -> None: + """Subscribe a handler to an event class.""" + self._subscribers[event_class].append(handler) + logger.debug(f"Subscribed handler to {event_class.__name__} events") + + def unsubscribe(self, handler: Handler) -> None: + """Remove a handler from all subscriptions.""" + for event_class in self._subscribers: + if handler in self._subscribers[event_class]: + self._subscribers[event_class].remove(handler) + logger.debug(f"Unsubscribed handler from {event_class.__name__} events") + + async def publish(self, event: Event) -> None: + """Publish an event to the internal queue (non-blocking).""" + await self._queue.put(event) + logger.debug(f"Queued {event.__class__.__name__} event from {event.source}") + + async def run(self) -> None: + """Process events from queue, starting with recovery.""" + logger.info("EventBus started") + + # Run recovery first + await self._recover() + + try: + while True: + event = await self._queue.get() + try: + await self._dispatch(event) + except Exception as e: + logger.error(f"Error dispatching event: {e}") + finally: + self._queue.task_done() + except asyncio.CancelledError: + logger.info("EventBus stopping...") + raise + + async def _dispatch(self, event: Event) -> None: + """Persist if OUTBOUND, then notify subscribers.""" + await self._persist_outbound(event) + await self._notify_subscribers(event) + logger.debug(f"Dispatched {event.__class__.__name__} event from {event.source}") + + async def _notify_subscribers(self, event: Event) -> None: + """Notify all subscribers of an event (waits for all handlers to complete).""" + handlers = self._subscribers.get(type(event), []) + if not handlers: + return + + tasks = [handler(event) for handler in handlers] + results = await asyncio.gather(*tasks, return_exceptions=True) + for result in results: + if isinstance(result, Exception): + logger.error(f"Error in event handler: {result}") + + async def _persist_outbound(self, event: Event) -> None: + """Persist event to disk (only OUTBOUND events).""" + if not isinstance(event, OutboundEvent): + return + + filename = f"{event.timestamp}_{event.session_id}.json" + final_path = self.pending_dir / filename + tmp_path = self.pending_dir / f".tmp.{os.getpid()}.{filename}" + + data = json.dumps(event.to_dict(), ensure_ascii=False) + + # Atomic write: tmp + fsync + rename + with open(tmp_path, "w", encoding="utf-8") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + + os.replace(str(tmp_path), str(final_path)) + logger.debug(f"Persisted event to {final_path}") + + async def _recover(self) -> int: + """Recover pending events from previous crash. Returns count recovered.""" + pending_files = list(self.pending_dir.glob("*.json")) + if not pending_files: + return 0 + + logger.info(f"Recovering {len(pending_files)} pending events") + count = 0 + + for file_path in pending_files: + try: + with open(file_path, "r", encoding="utf-8") as f: + data = json.load(f) + # Use deserialize_event to handle typed events + event = deserialize_event(data) + await self._notify_subscribers(event) + count += 1 + logger.debug(f"Recovered event from {file_path.name}") + except Exception as e: + logger.error(f"Failed to recover {file_path}: {e}") + + logger.info(f"Recovered {count} events") + return count + + def ack(self, event: Event) -> None: + """Acknowledge successful delivery, delete persisted event.""" + filename = f"{event.timestamp}_{event.session_id}.json" + final_path = self.pending_dir / filename + if final_path.exists(): + final_path.unlink() + logger.debug(f"Acked and deleted {filename}") diff --git a/12-cron-heartbeat/src/mybot/core/events.py b/12-cron-heartbeat/src/mybot/core/events.py new file mode 100644 index 0000000..1efc2f5 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/events.py @@ -0,0 +1,211 @@ +"""Event types and data classes for the event bus.""" + +import time +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import Any, ClassVar + + +class EventSource(ABC): + """Abstract base for all event sources.""" + + _registry: ClassVar[dict[str, type["EventSource"]]] = {} + _namespace: ClassVar[str] = "" + + def __init_subclass__(cls, **kwargs): + super().__init_subclass__(**kwargs) + if cls._namespace: + cls._registry[cls._namespace] = cls + + @property + def is_platform(self) -> bool: + return self._namespace.startswith("platform-") + + @property + def is_agent(self) -> bool: + return self._namespace == "agent" + + @property + def is_cron(self) -> bool: + return self._namespace == "cron" + + @property + def platform_name(self) -> str | None: + if not self.is_platform: + return None + return self._namespace.split("-", 1)[1] + + @classmethod + def from_string(cls, s: str) -> "EventSource": + """Parse string to EventSource using namespace registry.""" + namespace = s.split(":")[0] + source_cls = cls._registry.get(namespace) + if not source_cls: + raise ValueError(f"Unknown source namespace: {namespace}") + return source_cls.from_string(s) + + @abstractmethod + def __str__(self) -> str: ... + + +@dataclass +class AgentEventSource(EventSource): + """Source for agent-generated events.""" + + _namespace = "agent" + agent_id: str + + def __str__(self) -> str: + return f"agent:{self.agent_id}" + + @classmethod + def from_string(cls, s: str) -> "AgentEventSource": + _, agent_id = s.split(":", 1) + return cls(agent_id=agent_id) + + +@dataclass +class CliEventSource(EventSource): + """Source for CLI-originated events.""" + + _namespace = "platform-cli" + + def __str__(self) -> str: + return "platform-cli:cli-user" + + @classmethod + def from_string(cls, s: str) -> "CliEventSource": + return cls() + + @property + def platform_name(self) -> str: + return "cli" + + +@dataclass +class WebSocketEventSource(EventSource): + """Event from WebSocket client.""" + + _namespace = "platform-ws" + user_id: str + + @classmethod + def from_string(cls, s: str) -> "WebSocketEventSource": + """Parse source string into WebSocketEventSource.""" + parts = s.split(":", 1) + if len(parts) != 2 or parts[0] != cls._namespace or not parts[1]: + raise ValueError(f"Invalid WebSocketEventSource: {s}") + return cls(user_id=parts[1]) + + def __str__(self) -> str: + """Convert to source string format.""" + return f"{self._namespace}:{self.user_id}" + + @property + def is_platform(self) -> bool: + """WebSocket sources are platform sources.""" + return True + + +@dataclass +class CronEventSource(EventSource): + """Source for cron-triggered events.""" + + _namespace = "cron" + cron_id: str + + def __str__(self) -> str: + return f"cron:{self.cron_id}" + + @classmethod + def from_string(cls, s: str) -> "CronEventSource": + _, cron_id = s.split(":", 1) + return cls(cron_id=cron_id) + + +@dataclass +class Event: + """Base class for all typed events.""" + + session_id: str + source: EventSource # Changed from str to typed EventSource + content: str + timestamp: float = field(default_factory=time.time) + + def to_dict(self) -> dict[str, Any]: + """Serialize event to dictionary, including type.""" + result: dict[str, Any] = {"type": self.__class__.__name__} + for field_name in self.__dataclass_fields__: + value = getattr(self, field_name) + if field_name == "source": + result[field_name] = str(value) + else: + result[field_name] = value + return result + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "Event": + """Deserialize event from dictionary.""" + kwargs = {} + for k, v in data.items(): + if k == "type": + continue + if k == "source": + kwargs[k] = EventSource.from_string(v) + elif k in cls.__dataclass_fields__: + kwargs[k] = v + return cls(**kwargs) + + +@dataclass +class InboundEvent(Event): + """Event for external work entering the system (platforms, cron, retry).""" + + retry_count: int = 0 + + +@dataclass +class OutboundEvent(Event): + """Event for agent responses to deliver to platforms.""" + + error: str | None = None + + +@dataclass +class DispatchEvent(Event): + """Event for internal agent-to-agent delegation.""" + + parent_session_id: str = "" + retry_count: int = 0 + + +@dataclass +class DispatchResultEvent(Event): + """Event for result of a dispatched job.""" + + error: str | None = None + + +# Registry mapping event class names to event classes +_EVENT_CLASSES: dict[str, type[Event]] = { + "InboundEvent": InboundEvent, + "OutboundEvent": OutboundEvent, + "DispatchEvent": DispatchEvent, + "DispatchResultEvent": DispatchResultEvent, +} + + +def serialize_event(event: Event) -> dict[str, Any]: + """Serialize any event type to dict.""" + return event.to_dict() + + +def deserialize_event(data: dict[str, Any]) -> Event: + """Deserialize dict to appropriate event type.""" + event_type: str = data.get("type", "") + + event_class = _EVENT_CLASSES.get(event_type) + if event_class is None: + raise ValueError(f"Unknown event type: {event_type}") + + return event_class.from_dict(data) diff --git a/12-cron-heartbeat/src/mybot/core/history.py b/12-cron-heartbeat/src/mybot/core/history.py new file mode 100644 index 0000000..d47ca26 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/history.py @@ -0,0 +1,230 @@ +"""JSONL file-based conversation history backend.""" + +from datetime import datetime +from pathlib import Path +from typing import Any, Literal, TYPE_CHECKING + +from pydantic import BaseModel, Field, field_validator + +from mybot.core.events import EventSource +from litellm.types.completion import ChatCompletionMessageParam as Message + + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +def _now_iso() -> str: + """Return current datetime as ISO format string.""" + return datetime.now().isoformat() + + +class HistorySession(BaseModel): + """Session metadata - stored in index.jsonl.""" + + id: str + agent_id: str + source: str # Serialized EventSource (e.g., "platform-telegram:123:456") + title: str | None = None + message_count: int = 0 + created_at: str + updated_at: str + + @field_validator("source", mode="before") + @classmethod + def parse_source(cls, v: Any) -> str: + if hasattr(v, "__str__"): + return str(v) + return v + + def get_source(self) -> EventSource: + """Get the session's EventSource.""" + return EventSource.from_string(self.source) + + +class HistoryMessage(BaseModel): + """Single message - stored in session.jsonl.""" + + timestamp: str = Field(default_factory=_now_iso) + role: Literal["user", "assistant", "system", "tool"] + content: str + tool_calls: list[dict[str, Any]] | None = None + tool_call_id: str | None = None + + @classmethod + def from_message(cls, message: Message) -> "HistoryMessage": + """Create HistoryMessage from litellm Message format.""" + tool_calls = None + if message.get("tool_calls"): + tool_calls = [ + { + "id": tc.get("id"), + "type": tc.get("type", "function"), + "function": tc.get("function", {}), + } + for tc in message["tool_calls"] + ] + + tool_call_id = message.get("tool_call_id") + + return cls( + role=message["role"], + content=str(message.get("content", "")), + tool_calls=tool_calls, + tool_call_id=tool_call_id, + ) + + def to_message(self) -> Message: + """Convert HistoryMessage to litellm Message format.""" + base: dict[str, Any] = { + "role": self.role, + "content": self.content, + } + + if self.role == "assistant" and self.tool_calls: + return { + "role": "assistant", + "content": self.content, + "tool_calls": self.tool_calls, + } + + if self.role == "tool" and self.tool_call_id: + base["tool_call_id"] = self.tool_call_id + return base + + return base + + +class HistoryStore: + """JSONL file-based history storage.""" + + @staticmethod + def from_config(config: "Config") -> "HistoryStore": + return HistoryStore(config.history_path) + + def __init__(self, base_path: Path): + self.base_path = Path(base_path) + self.sessions_path = self.base_path / "sessions" + self.index_path = self.base_path / "index.jsonl" + + self.base_path.mkdir(parents=True, exist_ok=True) + self.sessions_path.mkdir(parents=True, exist_ok=True) + + def _session_path(self, session_id: str) -> Path: + """Get the file path for a session.""" + return self.sessions_path / f"{session_id}.jsonl" + + def _read_index(self) -> list[HistorySession]: + """Read all session entries from index.jsonl.""" + if not self.index_path.exists(): + return [] + + sessions = [] + with open(self.index_path) as f: + for line in f: + line = line.strip() + if line: + try: + sessions.append(HistorySession.model_validate_json(line)) + except Exception: + continue + return sessions + + def _write_index(self, sessions: list[HistorySession]) -> None: + """Write all session entries to index.jsonl.""" + with open(self.index_path, "w") as f: + for session in sessions: + f.write(session.model_dump_json() + "\n") + + def _find_session_index( + self, sessions: list[HistorySession], session_id: str + ) -> int: + """Find the index of a session in the list.""" + for i, s in enumerate(sessions): + if s.id == session_id: + return i + return -1 + + def create_session( + self, agent_id: str, session_id: str, source: "EventSource", + ) -> dict[str, Any]: + """Create a new conversation session.""" + now = _now_iso() + session = HistorySession( + id=session_id, + agent_id=agent_id, + source=source, + title=None, + message_count=0, + created_at=now, + updated_at=now, + ) + + # Append to index + with open(self.index_path, "a") as f: + f.write(session.model_dump_json() + "\n") + + # Create session file + self._session_path(session_id).touch() + + return session.model_dump() + + def save_message(self, session_id: str, message: HistoryMessage) -> None: + """Save a message to history.""" + sessions = self._read_index() + idx = self._find_session_index(sessions, session_id) + if idx < 0: + raise ValueError(f"Session not found: {session_id}") + + session = sessions[idx] + + # Append message to session file + session_file = self._session_path(session_id) + with open(session_file, "a") as f: + f.write(message.model_dump_json() + "\n") + + # Update index + session.message_count += 1 + session.updated_at = _now_iso() + + # Auto-generate title from first user message + if session.title is None and message.role == "user": + title = message.content[:50] + if len(message.content) > 50: + title += "..." + session.title = title + + sessions.sort(key=lambda s: s.updated_at, reverse=True) + self._write_index(sessions) + + def list_sessions(self) -> list[HistorySession]: + """List all sessions, most recently updated first.""" + sessions = self._read_index() + sessions.sort(key=lambda s: s.updated_at, reverse=True) + return sessions + + def get_messages(self, session_id: str) -> list[HistoryMessage]: + """Get all messages for a session.""" + session_file = self._session_path(session_id) + if not session_file.exists(): + return [] + + messages: list[HistoryMessage] = [] + with open(session_file) as f: + for line in f: + line = line.strip() + if line: + try: + messages.append(HistoryMessage.model_validate_json(line)) + except Exception: + continue + + return messages + + def get_session_info(self, session_id: str) -> HistorySession | None: + """Get session metadata without loading messages.""" + sessions = self._read_index() + for session in sessions: + if session.id == session_id: + return session + return None diff --git a/12-cron-heartbeat/src/mybot/core/routing.py b/12-cron-heartbeat/src/mybot/core/routing.py new file mode 100644 index 0000000..fbed1bd --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/routing.py @@ -0,0 +1,113 @@ +# src/mybot/core/routing.py + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from re import Pattern +from typing import TYPE_CHECKING + +from mybot.core.agent import Agent +from mybot.core.events import EventSource +from mybot.utils.config import SourceSessionConfig + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + + +@dataclass +class Binding: + """A routing binding that matches sources to agents.""" + + agent: str + value: str + tier: int = field(init=False) + pattern: Pattern = field(init=False) + + def __post_init__(self): + self.pattern = re.compile(f"^{self.value}$") + self.tier = self._compute_tier() + + def _compute_tier(self) -> int: + """Compute specificity tier.""" + if not any(c in self.value for c in r".*+?[]()|^$"): + return 0 + if ".*" in self.value: + return 2 + return 1 + + +@dataclass +class RoutingTable: + """Routes sources to agents using regex bindings.""" + + context: SharedContext + bindings: list[Binding] | None = field(default=None, init=False) + _config_hash: int | None = field(default=None, init=False) + + def _load_bindings(self) -> list[Binding]: + """Load and sort bindings from config. Cached until config changes.""" + bindings_data = self.context.config.routing.get("bindings", []) + current_hash = hash(tuple((b["agent"], b["value"]) for b in bindings_data)) + + if self.bindings is not None and self._config_hash == current_hash: + return self.bindings + + # Rebuild + bindings_with_order = [ + (Binding(agent=b["agent"], value=b["value"]), i) + for i, b in enumerate(bindings_data) + ] + bindings_with_order.sort(key=lambda x: (x[0].tier, x[1])) + self.bindings = [b for b, _ in bindings_with_order] + self._config_hash = current_hash + + return self.bindings + + def resolve(self, source: str) -> str: + """Return agent_id for source, falling back to default_agent if no match.""" + for binding in self._load_bindings(): + if binding.pattern.match(source): + return binding.agent + return self.context.config.default_agent + + def get_or_create_session_id(self, source: EventSource) -> str: + """Get existing or create new session_id for source.""" + source_str = str(source) + + source_session = self.context.config.sources.get(source_str) + if source_session: + return source_session.session_id + + agent_id = self.resolve(source_str) + agent_def = self.context.agent_loader.load(agent_id) + agent = Agent(agent_def, self.context) + session = agent.new_session(source) + + # Cache the session + self.context.config.set_runtime( + f"sources.{source_str}", SourceSessionConfig(session_id=session.session_id) + ) + + return session.session_id + + def persist_binding(self, source_pattern: str, agent_id: str) -> None: + """Add and persist a routing binding to config.user.yaml.""" + bindings = self.context.config.routing.get("bindings", []) + bindings.append({"agent": agent_id, "value": source_pattern}) + self.context.config.set_runtime("routing.bindings", bindings) + + def config_source_session_cache( + self, source_str: str, session_id: str | None + ) -> None: + """Config session cache for a source.""" + if session_id is None: + if source_str in self.context.config.sources: + del self.context.config.sources[source_str] + self.context.config.set_runtime( + "sources", self.context.config.sources + ) + else: + self.context.config.set_runtime( + f"""sources.{source_str}""", SourceSessionConfig(session_id=session_id) + ) \ No newline at end of file diff --git a/12-cron-heartbeat/src/mybot/core/session_state.py b/12-cron-heartbeat/src/mybot/core/session_state.py new file mode 100644 index 0000000..4289d08 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/session_state.py @@ -0,0 +1,37 @@ +"""Session state container with persistence helpers.""" + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from litellm.types.completion import ChatCompletionMessageParam as Message + +from mybot.core.history import HistoryMessage + +if TYPE_CHECKING: + from mybot.core.agent import Agent + from mybot.core.context import SharedContext + from mybot.core.events import EventSource + + +@dataclass +class SessionState: + """Pure conversation state + persistence.""" + + session_id: str + agent: "Agent" + messages: list[Message] + source: "EventSource" + shared_context: "SharedContext" + + def add_message(self, message: Message) -> None: + """Add message to in-memory list + persist.""" + self.messages.append(message) + history_msg = HistoryMessage.from_message(message) + self.shared_context.history_store.save_message(self.session_id, history_msg) + + def build_messages(self) -> list[Message]: + """Build messages list with system prompt.""" + system_prompt = self.agent.agent_def.agent_md + messages: list[Message] = [{"role": "system", "content": system_prompt}] + messages.extend(self.messages) + return messages diff --git a/12-cron-heartbeat/src/mybot/core/skill_loader.py b/12-cron-heartbeat/src/mybot/core/skill_loader.py new file mode 100644 index 0000000..a590ff1 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/core/skill_loader.py @@ -0,0 +1,69 @@ +"""Skill loader for discovering and loading skills.""" + +import logging +from typing import TYPE_CHECKING, Any + +from pydantic import BaseModel, ConfigDict, ValidationError + +from mybot.utils.def_loader import DefNotFoundError, discover_definitions + +if TYPE_CHECKING: + from mybot.utils.config import Config + +logger = logging.getLogger(__name__) + + +class SkillDef(BaseModel): + """Loaded skill definition.""" + + model_config = ConfigDict(extra="forbid") + + id: str + name: str + description: str + content: str + + +class SkillLoader: + """Load and manage skill definitions from filesystem.""" + + @staticmethod + def from_config(config: "Config") -> "SkillLoader": + """Create SkillLoader from config.""" + return SkillLoader(config) + + def __init__(self, config: "Config"): + self.config = config + + def discover_skills(self) -> list[SkillDef]: + """Scan skills directory and return list of valid SkillDef.""" + return discover_definitions( + self.config.skills_path, "SKILL.md", self._parse_skill_def + ) + + def _parse_skill_def( + self, def_id: str, frontmatter: dict[str, Any], body: str + ) -> SkillDef | None: + """Parse skill definition from frontmatter (callback for discover_definitions).""" + try: + return SkillDef( + id=def_id, + name=frontmatter["name"], # type: ignore[misc] + description=frontmatter["description"], # type: ignore[misc] + content=body.strip(), + ) + except ValidationError as e: + logger.warning(f"Invalid skill '{def_id}': {e}") + return None + except KeyError as e: + logger.warning(f"Missing required field in skill '{def_id}': {e}") + return None + + def load_skill(self, skill_id: str) -> SkillDef: + """Load full skill definition by ID.""" + skills = self.discover_skills() + for skill in skills: + if skill.id == skill_id: + return skill + + raise DefNotFoundError("skill", skill_id) diff --git a/12-cron-heartbeat/src/mybot/provider/__init__.py b/12-cron-heartbeat/src/mybot/provider/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/12-cron-heartbeat/src/mybot/provider/llm/__init__.py b/12-cron-heartbeat/src/mybot/provider/llm/__init__.py new file mode 100644 index 0000000..c1dfcd6 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/llm/__init__.py @@ -0,0 +1,5 @@ +"""LLM provider abstraction.""" + +from .base import LLMProvider, LLMToolCall + +__all__ = ["LLMProvider", "LLMToolCall"] diff --git a/12-cron-heartbeat/src/mybot/provider/llm/base.py b/12-cron-heartbeat/src/mybot/provider/llm/base.py new file mode 100644 index 0000000..8f52788 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/llm/base.py @@ -0,0 +1,85 @@ +"""Base LLM provider abstraction.""" + +from dataclasses import dataclass +from typing import Any, Optional, cast + +from litellm import acompletion, Choices +from litellm.types.completion import ChatCompletionMessageParam as Message + +from mybot.utils.config import LLMConfig + + +@dataclass +class LLMToolCall: + """A tool/function call from the LLM.""" + + id: str + name: str + arguments: str # JSON string + + +class LLMProvider: + """LLM provider using litellm for multi-provider support.""" + + def __init__( + self, + model: str, + api_key: str, + api_base: Optional[str] = None, + temperature: float = 0.7, + max_tokens: int = 2048, + **kwargs: Any, + ): + """Initialize LLM provider.""" + self.model = model + self.api_key = api_key + self.api_base = api_base + self.temperature = temperature + self.max_tokens = max_tokens + self._settings = kwargs + + @classmethod + def from_config(cls, config: LLMConfig) -> "LLMProvider": + """Create provider from LLMConfig.""" + return cls( + model=config.model, + api_key=config.api_key, + api_base=config.api_base, + temperature=config.temperature, + max_tokens=config.max_tokens, + ) + + async def chat( + self, + messages: list[Message], + tools: Optional[list[dict[str, Any]]] = None, + **kwargs: Any, + ) -> tuple[str, list[LLMToolCall]]: + """Default implementation using litellm. Subclasses can override.""" + request_kwargs: dict[str, Any] = { + "model": self.model, + "messages": messages, + "api_key": self.api_key, + } + + if self.api_base: + request_kwargs["api_base"] = self.api_base + if tools: + request_kwargs["tools"] = tools + request_kwargs.update(kwargs) + + response = await acompletion(**request_kwargs) + + message = cast(Choices, response.choices[0]).message + + return ( + message.content or "", + [ + LLMToolCall( + id=tc["id"], + name=tc["function"]["name"], + arguments=tc["function"]["arguments"], + ) + for tc in (message.tool_calls or []) + ], + ) diff --git a/12-cron-heartbeat/src/mybot/provider/web_read/__init__.py b/12-cron-heartbeat/src/mybot/provider/web_read/__init__.py new file mode 100644 index 0000000..3d8da67 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/web_read/__init__.py @@ -0,0 +1,5 @@ +"""Web read provider module.""" + +from .base import ReadResult, WebReadProvider + +__all__ = ["ReadResult", "WebReadProvider"] diff --git a/12-cron-heartbeat/src/mybot/provider/web_read/base.py b/12-cron-heartbeat/src/mybot/provider/web_read/base.py new file mode 100644 index 0000000..ce410e2 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/web_read/base.py @@ -0,0 +1,41 @@ +"""Base class for web read providers.""" + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from pydantic import BaseModel + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +class ReadResult(BaseModel): + """Normalized result from reading a web page.""" + + url: str + title: str + content: str # Markdown content + error: str | None = None + + +class WebReadProvider(ABC): + """Abstract base class for web page reading providers.""" + + @abstractmethod + async def read(self, url: str) -> ReadResult: + """Read a web page and return normalized content.""" + pass + + @staticmethod + def from_config(config: "Config") -> "WebReadProvider": + """Factory method to create provider from config.""" + if config.webread is None: + raise ValueError("Webread not configured") + + match config.webread.provider: + case "crawl4ai": + from .crawl4ai import Crawl4AIProvider + + return Crawl4AIProvider() + case _: + raise ValueError(f"Unknown webread provider: {config.webread.provider}") diff --git a/12-cron-heartbeat/src/mybot/provider/web_read/crawl4ai.py b/12-cron-heartbeat/src/mybot/provider/web_read/crawl4ai.py new file mode 100644 index 0000000..4feb85f --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/web_read/crawl4ai.py @@ -0,0 +1,36 @@ +"""Crawl4AI provider for web page reading.""" + +from crawl4ai import AsyncWebCrawler + +from .base import WebReadProvider, ReadResult + + +class Crawl4AIProvider(WebReadProvider): + """Web read provider using Crawl4AI.""" + + def __init__(self): + """Initialize Crawl4AI provider.""" + pass + + async def read(self, url: str) -> ReadResult: + """Read a web page using Crawl4AI.""" + try: + async with AsyncWebCrawler(verbose=False) as crawler: + result = await crawler.arun(url=url) + + if not result.success: + raise Exception(result.error_message or "Failed to crawl page") + + return ReadResult( + url=url, + title=(result.metadata.get("title", "") if result.metadata else ""), + content=result.markdown or "", + error=None, + ) + except Exception as e: + return ReadResult( + url=url, + title="", + content="", + error=str(e), + ) diff --git a/12-cron-heartbeat/src/mybot/provider/web_search/__init__.py b/12-cron-heartbeat/src/mybot/provider/web_search/__init__.py new file mode 100644 index 0000000..38761eb --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/web_search/__init__.py @@ -0,0 +1,5 @@ +"""Web search provider module.""" + +from .base import SearchResult, WebSearchProvider + +__all__ = ["SearchResult", "WebSearchProvider"] diff --git a/12-cron-heartbeat/src/mybot/provider/web_search/base.py b/12-cron-heartbeat/src/mybot/provider/web_search/base.py new file mode 100644 index 0000000..e5793c4 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/web_search/base.py @@ -0,0 +1,42 @@ +"""Base class for web search providers.""" + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from pydantic import BaseModel + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +class SearchResult(BaseModel): + """Normalized search result from any provider.""" + + title: str + url: str + snippet: str + + +class WebSearchProvider(ABC): + """Abstract base class for web search providers.""" + + @abstractmethod + async def search(self, query: str) -> list[SearchResult]: + """Search the web and return normalized results.""" + pass + + @staticmethod + def from_config(config: "Config") -> "WebSearchProvider": + """Factory method to create provider from config.""" + if config.websearch is None: + raise ValueError("Websearch not configured") + + match config.websearch.provider: + case "brave": + from .brave import BraveSearchProvider + + return BraveSearchProvider(config) + case _: + raise ValueError( + f"Unknown websearch provider: {config.websearch.provider}" + ) diff --git a/12-cron-heartbeat/src/mybot/provider/web_search/brave.py b/12-cron-heartbeat/src/mybot/provider/web_search/brave.py new file mode 100644 index 0000000..1941e48 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/provider/web_search/brave.py @@ -0,0 +1,49 @@ +"""Brave Search API provider.""" + +from typing import TYPE_CHECKING +import httpx + +from .base import WebSearchProvider, SearchResult + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +class BraveSearchProvider(WebSearchProvider): + """Web search provider using Brave Search API.""" + + BASE_URL = "https://api.search.brave.com/res/v1/web/search" + + def __init__(self, config: "Config"): + """Initialize Brave Search provider.""" + self.api_key = config.websearch.api_key + + async def search(self, query: str) -> list[SearchResult]: + """Search the web using Brave Search API.""" + async with httpx.AsyncClient() as client: + response = await client.get( + self.BASE_URL, + headers={ + "Accept": "application/json", + "X-Subscription-Token": self.api_key, + }, + params={ + "q": query, + "count": 10, + }, + timeout=30.0, + ) + response.raise_for_status() + data = response.json() + + results = [] + for item in data.get("web", {}).get("results", []): + results.append( + SearchResult( + title=item.get("title", ""), + url=item.get("url", ""), + snippet=item.get("description", ""), + ) + ) + + return results diff --git a/12-cron-heartbeat/src/mybot/server/__init__.py b/12-cron-heartbeat/src/mybot/server/__init__.py new file mode 100644 index 0000000..95dc6fe --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/__init__.py @@ -0,0 +1,18 @@ +"""Worker-based server architecture.""" + +from .worker import Worker, SubscriberWorker +from .delivery_worker import DeliveryWorker +from .websocket_worker import WebSocketWorker +from .agent_worker import AgentWorker +from .cron_worker import CronWorker +from .channel_worker import ChannelWorker + +__all__ = [ + "Worker", + "SubscriberWorker", + "DeliveryWorker", + "WebSocketWorker", + "AgentWorker", + "CronWorker", + "ChannelWorker", +] diff --git a/12-cron-heartbeat/src/mybot/server/agent_worker.py b/12-cron-heartbeat/src/mybot/server/agent_worker.py new file mode 100644 index 0000000..dd2ad12 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/agent_worker.py @@ -0,0 +1,126 @@ +"""Agent worker for executing agent jobs.""" + +import asyncio +import logging +from dataclasses import replace +from typing import Union + +from .worker import SubscriberWorker +from mybot.core.agent import Agent +from mybot.core.events import ( + AgentEventSource, + InboundEvent, + OutboundEvent, + DispatchEvent, + DispatchResultEvent, +) +from mybot.utils.def_loader import DefNotFoundError + + +# Maximum number of retry attempts for failed sessions +MAX_RETRIES = 3 + +logger = logging.getLogger(__name__) + +ProcessEvent = Union[InboundEvent, DispatchEvent] + + +class AgentWorker(SubscriberWorker): + """Dispatches events to session executors.""" + + def __init__(self, context): + super().__init__(context) + + # Auto-subscribe to events + self.context.eventbus.subscribe(InboundEvent, self.dispatch_event) + self.context.eventbus.subscribe(DispatchEvent, self.dispatch_event) + self.logger.info("AgentWorker subscribed to InboundEvent and DispatchEvent events") + + async def dispatch_event(self, event: ProcessEvent) -> None: + """Create executor task for typed event.""" + # Get agent_id from session (single source of truth) + session_info = self.context.history_store.get_session_info(event.session_id) + if not session_info: + logger.error(f"Session not found: {event.session_id}") + return + + agent_id = session_info.agent_id + + try: + agent_def = self.context.agent_loader.load(agent_id) + except DefNotFoundError as e: + logger.error(f"Agent not found: {agent_id}: {e}") + + return await self._emit_response(event, "", agent_def.id, str(e)) + + asyncio.create_task(self.exec_session(event, agent_def)) + + async def exec_session(self, event: ProcessEvent, agent_def) -> None: + session_id = event.session_id + + try: + agent = Agent(agent_def, self.context) + if session_id: + try: + session = agent.resume_session(session_id) + except ValueError: + logger.warning(f"Session {session_id} not found, creating new") + session = agent.new_session(session_id=session_id) + else: + session = agent.new_session() + session_id = session.session_id + + # Check for slash command FIRST + if event.content.startswith("/"): + result = await self.context.command_registry.dispatch( + event.content, session + ) + if result: + # Emit response and skip agent chat + await self._emit_response(event, result, agent_def.id) + logger.info(f"Command completed: {session_id}") + return + + response = await session.chat(event.content) + logger.info(f"Session completed: {session_id}") + + await self._emit_response(event, response, agent_def.id) + + except Exception as e: + logger.error(f"Session failed: {e}") + + if event.retry_count < MAX_RETRIES: + # Use dataclasses.replace() for retry logic + retry_event = replace( + event, + retry_count=event.retry_count + 1, + content=".", # Minimal message for retry + ) + await self.context.eventbus.publish(retry_event) + else: + await self._emit_response(event, "", agent_def.id, str(e)) + + + async def _emit_response( + self, + event: ProcessEvent, + content: str, + agent_id: str, + error: str | None = None, + ) -> None: + """Emit response event with content.""" + if isinstance(event, DispatchEvent): + result_event: OutboundEvent | DispatchResultEvent = DispatchResultEvent( + session_id=event.session_id, + source=AgentEventSource(agent_id), + content=content, + error=str(error) if error else None, + ) + else: + result_event = OutboundEvent( + session_id=event.session_id, + source=AgentEventSource(agent_id), + content=content, + error=str(error) if error else None, + ) + await self.context.eventbus.publish(result_event) diff --git a/12-cron-heartbeat/src/mybot/server/app.py b/12-cron-heartbeat/src/mybot/server/app.py new file mode 100644 index 0000000..9184a42 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/app.py @@ -0,0 +1,41 @@ +"""FastAPI application with WebSocket support.""" + +from fastapi import FastAPI, WebSocket +from fastapi.middleware.cors import CORSMiddleware + +from mybot.core.context import SharedContext + + +def create_app(context: SharedContext) -> FastAPI: + """Create and configure the FastAPI application.""" + app = FastAPI( + title="MyBot WebSocket Server", + description="WebSocket server for real-time agent communication", + version="0.1.0", + ) + app.state.context = context + + # Enable CORS for web clients + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], + ) + + + # WebSocket endpoint + @app.websocket("/ws") + async def websocket_endpoint(websocket: WebSocket): + """WebSocket endpoint for real-time event streaming and chat.""" + await websocket.accept() + + # Check if WebSocket worker is available + if context.websocket_worker is None: + await websocket.close(code=1013, reason="WebSocket not available") + return + + # Hand off to worker + await context.websocket_worker.handle_connection(websocket) + + return app diff --git a/12-cron-heartbeat/src/mybot/server/channel_worker.py b/12-cron-heartbeat/src/mybot/server/channel_worker.py new file mode 100644 index 0000000..7ec8d14 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/channel_worker.py @@ -0,0 +1,73 @@ +"""Channel worker for ingesting platform messages.""" + +import asyncio +import time +from typing import TYPE_CHECKING + +from .worker import Worker +from mybot.core.events import EventSource, InboundEvent + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + + +class ChannelWorker(Worker): + """Ingests messages from platforms, publishes INBOUND events to Channel.""" + + def __init__(self, context: "SharedContext"): + super().__init__(context) + self.channels = context.channels + self.channel_map = {channel.platform_name: channel for channel in self.channels} + + async def run(self) -> None: + """Start all channels and process incoming messages.""" + self.logger.info(f"ChannelWorker started with {len(self.channels)} channel(es)") + + channel_tasks = [ + channel.run(self._create_callback(channel.platform_name)) + for channel in self.channels + ] + + try: + await asyncio.gather(*channel_tasks) + except asyncio.CancelledError: + await asyncio.gather(*[channel.stop() for channel in self.channels]) + raise + + def _create_callback(self, platform: str): + """Create callback for a specific platform.""" + + async def callback(message: str, source: EventSource) -> None: + try: + channel = self.channel_map[platform] + + if not channel.is_allowed(source): + self.logger.debug( + f"Ignored non-whitelisted message from {platform}" + ) + return + + # Set default delivery source only on first non-CLI platform message + if source.is_platform and source.platform_name != "cli": + if not self.context.config.default_delivery_source: + source_str_value = str(source) + self.context.config.set_runtime( + "default_delivery_source", source_str_value + ) + + session_id = self.context.routing_table.get_or_create_session_id(source) + + # Publish INBOUND event with typed source + event = InboundEvent( + session_id=session_id, + source=source, + content=message, + timestamp=time.time(), + ) + await self.context.eventbus.publish(event) + self.logger.debug(f"Published INBOUND event from {source}") + + except Exception as e: + self.logger.error(f"Error processing message from {platform}: {e}") + + return callback \ No newline at end of file diff --git a/12-cron-heartbeat/src/mybot/server/cron_worker.py b/12-cron-heartbeat/src/mybot/server/cron_worker.py new file mode 100644 index 0000000..e1a0e6a --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/cron_worker.py @@ -0,0 +1,84 @@ +"""Cron worker for scheduled job dispatch.""" + +import asyncio +import logging +import shutil +from datetime import datetime +from typing import TYPE_CHECKING + +from croniter import croniter + +from .worker import Worker +from mybot.core.agent import Agent +from mybot.core.events import CronEventSource, DispatchEvent + +if TYPE_CHECKING: + from mybot.core.cron_loader import CronDef + from mybot.core.context import SharedContext + +logger = logging.getLogger(__name__) + + +def find_due_jobs( + jobs: list["CronDef"], now: datetime | None = None +) -> list["CronDef"]: + """Find all jobs that are due to run.""" + if not jobs: + return [] + + now = now or datetime.now() + now_minute = now.replace(second=0, microsecond=0) + + due_jobs = [] + for job in jobs: + try: + if croniter.match(job.schedule, now_minute): + due_jobs.append(job) + except Exception as e: + logger.warning(f"Error checking schedule for {job.id}: {e}") + continue + + return due_jobs + + +class CronWorker(Worker): + """Finds due cron jobs, publishes DISPATCH events.""" + + def __init__(self, context: "SharedContext"): + super().__init__(context) + + async def run(self) -> None: + """Check every minute for due jobs.""" + self.logger.info("CronWorker started") + + while True: + try: + await self._tick() + except Exception as e: + self.logger.error(f"Error in tick: {e}") + + await asyncio.sleep(60) + + async def _tick(self) -> None: + """Find and dispatch due jobs via EventBus.""" + jobs = self.context.cron_loader.discover_crons() + due_jobs = find_due_jobs(jobs) + + for cron_def in due_jobs: + agent_def = self.context.agent_loader.load(cron_def.agent) + agent = Agent(agent_def, self.context) + cron_source = CronEventSource(cron_id=cron_def.id) + session = agent.new_session(cron_source) + + event = DispatchEvent( + session_id=session.session_id, + source=CronEventSource(cron_id=cron_def.id), + content=cron_def.prompt, + ) + await self.context.eventbus.publish(event) + self.logger.info(f"Dispatched cron job: {cron_def.id}") + + if cron_def.one_off: + cron_path = self.context.cron_loader.config.crons_path / cron_def.id + shutil.rmtree(cron_path) + self.logger.info(f"Deleted one-off cron job: {cron_def.id}") diff --git a/12-cron-heartbeat/src/mybot/server/delivery_worker.py b/12-cron-heartbeat/src/mybot/server/delivery_worker.py new file mode 100644 index 0000000..d91bd7a --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/delivery_worker.py @@ -0,0 +1,194 @@ +"""Worker that delivers outbound messages to platforms.""" + +import asyncio +import logging +import random +from functools import lru_cache +from typing import TYPE_CHECKING, Any + +from mybot.core.events import EventSource, OutboundEvent +from mybot.core.history import HistorySession +from .worker import SubscriberWorker + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + from mybot.channel.base import Channel + +logger = logging.getLogger(__name__) + +# Retry configuration +BACKOFF_MS = [5000, 25000, 120000, 600000] # 5s, 25s, 2min, 10min +MAX_RETRIES = 5 + + +def compute_backoff_ms(retry_count: int) -> int: + """Compute backoff time with jitter.""" + if retry_count <= 0: + return 0 + + # Cap at last backoff value + idx = min(retry_count - 1, len(BACKOFF_MS) - 1) + base = BACKOFF_MS[idx] + + # Add +/- 20% jitter + jitter = random.randint(-base // 5, base // 5) + return max(0, base + jitter) + + +# Platform message size limits +PLATFORM_LIMITS: dict[str, float] = { + "telegram": 4096, + "discord": 2000, + "cli": float("inf"), # no limit +} + + +def chunk_message(content: str, limit: int) -> list[str]: + """Split message at paragraph boundaries, respecting limit.""" + if len(content) <= limit: + return [content] + + chunks = [] + paragraphs = content.split("\n\n") + current = "" + + for para in paragraphs: + # Try to add to current chunk + if current: + potential = current + "\n\n" + para + else: + potential = para + + if len(potential) <= limit: + current = potential + else: + if current: + chunks.append(current) + + # Handle paragraph that exceeds limit + if len(para) > limit: + # Hard split + for i in range(0, len(para), limit): + chunks.append(para[i : i + limit]) + current = "" + else: + current = para + + if current: + chunks.append(current) + + return chunks + + +class DeliveryWorker(SubscriberWorker): + """Worker that delivers outbound messages to platforms.""" + + def __init__(self, context: "SharedContext"): + super().__init__(context) + self.context.eventbus.subscribe(OutboundEvent, self.handle_event) + self.logger.info("DeliveryWorker subscribed to OUTBOUND events") + + async def _deliver_with_retry( + self, chunks: list[str], source: "EventSource", channel: "Channel[Any]" + ) -> bool: + """Deliver all chunks with retry logic. Returns True on success.""" + for attempt in range(1, MAX_RETRIES + 1): + try: + for chunk in chunks: + await channel.reply(chunk, source) + return True + except Exception as e: + if attempt < MAX_RETRIES: + backoff_ms = compute_backoff_ms(attempt) + self.logger.warning( + f"Delivery failed (attempt {attempt}/{MAX_RETRIES}), " + f"retrying in {backoff_ms}ms: {e}" + ) + await asyncio.sleep(backoff_ms / 1000) + else: + self.logger.error( + f"Delivery failed after {MAX_RETRIES} attempts: {e}" + ) + return False + return False + + @lru_cache(maxsize=10) + def _get_session_source(self, session_id: str) -> HistorySession | None: + """Get session info from HistoryStore (cached).""" + for session in self.context.history_store.list_sessions(): + if session.id == session_id: + return session + return None + + def _get_delivery_source( + self, session_info: HistorySession + ) -> "EventSource | None": + """Get the delivery source for a session.""" + source = session_info.get_source() + + # If source already has a platform, use it + if source.platform_name: + return source + + # Try default delivery source for agent events + default_source_str = self.context.config.default_delivery_source + if default_source_str: + try: + source = EventSource.from_string(default_source_str) + if not source.platform_name: + self.logger.error( + f"default_delivery_source '{default_source_str}' is not a platform source" + ) + return None + return source + except ValueError as e: + self.logger.error(f"Invalid default_delivery_source: {e}") + return None + else: + self.logger.warning( + f"No platform for session {session_info.id} and no default_delivery_source configured" + ) + return None + + async def handle_event(self, event: OutboundEvent) -> None: + """Handle an outbound message event.""" + try: + session_info = self._get_session_source(event.session_id) + + if not session_info or not session_info.source: + self.logger.warning( + f"No source for session {event.session_id}, skipping delivery" + ) + return + + source = self._get_delivery_source(session_info) + if not source or not source.platform_name: + # No valid delivery source - don't ack, let event be retried + return + + limit = PLATFORM_LIMITS.get(source.platform_name, float("inf")) + chunks = chunk_message( + event.content, + int(limit) if limit != float("inf") else len(event.content), + ) + + channel = self._get_channel(source.platform_name) + if channel: + success = await self._deliver_with_retry(chunks, source, channel) + if not success: + self.logger.error(f"Dropped message for session {event.session_id}") + + self.context.eventbus.ack(event) + self.logger.info( + f"Delivered message to {source.platform_name} for session {event.session_id}" + ) + + except Exception as e: + self.logger.error(f"Failed to deliver message: {e}") + + def _get_channel(self, platform: str) -> "Channel[Any] | None": + """Get the message channel for a platform.""" + for channel in self.context.channels: + if channel.platform_name == platform: + return channel + return None diff --git a/12-cron-heartbeat/src/mybot/server/server.py b/12-cron-heartbeat/src/mybot/server/server.py new file mode 100644 index 0000000..87b1ebd --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/server.py @@ -0,0 +1,123 @@ +"""Server orchestrator for worker-based architecture.""" + +import asyncio +import logging +from typing import TYPE_CHECKING + +import uvicorn + +from .worker import Worker +from .agent_worker import AgentWorker +from .cron_worker import CronWorker +from .delivery_worker import DeliveryWorker +from .channel_worker import ChannelWorker +from .websocket_worker import WebSocketWorker +from .app import create_app +from mybot.utils.config import ConfigReloader + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + +logger = logging.getLogger(__name__) + + +class Server: + """Orchestrates workers with queue-based communication.""" + + def __init__(self, context: "SharedContext"): + self.context = context + self.workers: list[Worker] = [] + self._api_task: asyncio.Task | None = None + self.config_reloader: ConfigReloader = ConfigReloader(self.context.config) + + async def run(self) -> None: + """Start all workers and monitor for crashes.""" + self._setup_workers() + self._start_workers() + + # Start API server if configured + if self.context.config.api: + self._api_task = asyncio.create_task(self._run_api()) + + try: + await self._monitor_workers() + except asyncio.CancelledError: + logger.info("Server shutting down...") + await self._stop_all() + raise + + def _setup_workers(self) -> None: + """Create all workers.""" + self.config_reloader.start() + + # Create WebSocketWorker first and attach to context + ws_worker = WebSocketWorker(self.context) + self.context.websocket_worker = ws_worker + + self.workers = [ + self.context.eventbus, # EventBus (active worker) + AgentWorker(self.context), # SubscriberWorker + DeliveryWorker(self.context), # SubscriberWorker + CronWorker(self.context), # Background worker for scheduled tasks + ws_worker, # WebSocketWorker (SubscriberWorker) + ] + + if self.context.config.channels.enabled: + channels = self.context.channels + if channels: + self.workers.append(ChannelWorker(self.context)) + logger.info(f"Channel enabled with {len(channels)} channel(es)") + else: + logger.warning("Channel enabled but no channels configured") + + logger.info(f"Server setup complete with {len(self.workers)} core workers") + + def _start_workers(self) -> None: + """Start all workers as tasks.""" + for worker in self.workers: + worker.start() + logger.info(f"Started {worker.__class__.__name__}") + + async def _monitor_workers(self) -> None: + """Monitor worker tasks, restart on crash.""" + while True: + for worker in self.workers: + if worker.has_crashed(): + exc = worker.get_exception() + if exc is None: + logger.warning( + f"{worker.__class__.__name__} exited unexpectedly" + ) + else: + logger.error(f"{worker.__class__.__name__} crashed: {exc}") + + worker.start() + logger.info(f"Restarted {worker.__class__.__name__}") + + await asyncio.sleep(5) + + async def _stop_all(self) -> None: + """Stop all workers gracefully.""" + for worker in self.workers: + await worker.stop() + + # Stop config reloader + if self.config_reloader is not None: + self.config_reloader.stop() + + async def _run_api(self) -> None: + """Run the WebSocket API server.""" + if not self.context.config.api: + return + + app = create_app(self.context) + config = uvicorn.Config( + app, + host=self.context.config.api.host, + port=self.context.config.api.port, + ) + server = uvicorn.Server(config) + logger.info( + f"WebSocket server started on {self.context.config.api.host}:{self.context.config.api.port}" + ) + await server.serve() diff --git a/12-cron-heartbeat/src/mybot/server/websocket_worker.py b/12-cron-heartbeat/src/mybot/server/websocket_worker.py new file mode 100644 index 0000000..74c5bdb --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/websocket_worker.py @@ -0,0 +1,138 @@ +"""WebSocket worker for broadcasting events to connected clients.""" + +import logging +import time +import dataclasses +from typing import TYPE_CHECKING, Set + + +from fastapi import WebSocket +from fastapi.websockets import WebSocketDisconnect +from pydantic import ValidationError, BaseModel, Field + +from .worker import SubscriberWorker +from mybot.core.events import ( + Event, + InboundEvent, + OutboundEvent, + DispatchEvent, + DispatchResultEvent, + WebSocketEventSource, +) +if TYPE_CHECKING: + from mybot.core.context import SharedContext + +logger = logging.getLogger(__name__) + + +class WebSocketMessage(BaseModel): + """Incoming WebSocket message from client.""" + + source: str = Field(..., min_length=1, description="Client identifier") + content: str = Field(..., min_length=1, description="Message content") + agent_id: str | None = Field( + None, description="Target agent ID (optional - uses routing if not specified)" + ) + + +class WebSocketWorker(SubscriberWorker): + """Manages WebSocket connections and event broadcasting.""" + + def __init__(self, context: "SharedContext"): + super().__init__(context) + self.clients: Set[WebSocket] = set() + + # Auto-subscribe to event classes + for event_class in [ + InboundEvent, + OutboundEvent, + DispatchEvent, + DispatchResultEvent + ]: + self.context.eventbus.subscribe(event_class, self.handle_event) + self.logger.info("WebSocketWorker subscribed to event types") + + async def handle_connection(self, ws: WebSocket) -> None: + """Handle a single WebSocket connection lifecycle.""" + self.clients.add(ws) + self.logger.info( + f"WebSocket client connected. Total clients: {len(self.clients)}" + ) + + try: + await self._run_client_loop(ws) + finally: + self.clients.discard(ws) + self.logger.info( + f"WebSocket client disconnected. Total clients: {len(self.clients)}" + ) + + async def _run_client_loop(self, ws: WebSocket) -> None: + """Run message receiving loop for a single client.""" + + while True: + try: + data = await ws.receive_json() + msg = WebSocketMessage(**data) + + event = self._normalize_message(msg) + + await self.context.eventbus.publish(event) + self.logger.debug(f"Emitted InboundEvent from WebSocket: {msg.source}") + + except WebSocketDisconnect: + self.logger.info("Client disconnected normally") + break + except ValidationError as e: + await ws.send_json( + {"type": "error", "message": f"Validation error: {e}"} + ) + self.logger.warning(f"Validation error from client: {e}") + except Exception as e: + self.logger.error(f"Unexpected error in client loop: {e}") + break + + def _normalize_message(self, msg: "WebSocketMessage") -> InboundEvent: + """Normalize WebSocketMessage to InboundEvent.""" + source = WebSocketEventSource(user_id=msg.source) + + agent_id = msg.agent_id + if agent_id is None: + agent_id = self.context.routing_table.resolve(str(source)) + + session_id = self.context.routing_table.get_or_create_session_id(source) + + return InboundEvent( + session_id=session_id, + source=source, + content=msg.content, + timestamp=time.time(), + ) + + async def handle_event(self, event: Event) -> None: + """Handle EventBus event by broadcasting to WebSocket clients.""" + if not self.clients: + return + + # Serialize event to dict with type information + event_dict = { + "type": event.__class__.__name__, + } + event_dict.update(dataclasses.asdict(event)) + + # Convert EventSource to string for JSON serialization + if "source" in event_dict and hasattr(event.source, "__str__"): + event_dict["source"] = str(event.source) + + # Broadcast to all clients + self.logger.debug( + f"Broadcasting {event.__class__.__name__} to {len(self.clients)} clients" + ) + + for client in list(self.clients): + try: + await client.send_json(event_dict) + except Exception as e: + self.logger.error(f"Failed to send to client: {e}") + self.clients.discard(client) + diff --git a/12-cron-heartbeat/src/mybot/server/worker.py b/12-cron-heartbeat/src/mybot/server/worker.py new file mode 100644 index 0000000..69593dd --- /dev/null +++ b/12-cron-heartbeat/src/mybot/server/worker.py @@ -0,0 +1,60 @@ +"""Base worker lifecycle management.""" + +import asyncio +import logging +from abc import ABC, abstractmethod + + +class Worker(ABC): + """Base class for all workers with lifecycle management.""" + + def __init__(self, context): + self.context = context + self.logger = logging.getLogger(f"mybot.server.{self.__class__.__name__}") + self._task: asyncio.Task | None = None + + @abstractmethod + async def run(self) -> None: + """Main worker loop. Runs until cancelled.""" + pass + + def start(self) -> asyncio.Task: + """Start the worker as an asyncio Task.""" + self._task = asyncio.create_task(self.run()) + return self._task + + def is_running(self) -> bool: + """Check if worker is actively running.""" + return self._task is not None and not self._task.done() + + def has_crashed(self) -> bool: + """Check if worker crashed (done but not cancelled).""" + return ( + self._task is not None and self._task.done() and not self._task.cancelled() + ) + + def get_exception(self) -> BaseException | None: + """Get the exception if worker crashed, None otherwise.""" + if self.has_crashed() and self._task is not None: + return self._task.exception() + return None + + async def stop(self) -> None: + """Gracefully stop the worker.""" + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + + +class SubscriberWorker(Worker): + """Worker that only subscribes to events, no active loop.""" + + async def run(self) -> None: + """Wait for cancellation - actual work happens in event handlers.""" + try: + await asyncio.Future() + except asyncio.CancelledError: + pass diff --git a/12-cron-heartbeat/src/mybot/tools/__init__.py b/12-cron-heartbeat/src/mybot/tools/__init__.py new file mode 100644 index 0000000..843290d --- /dev/null +++ b/12-cron-heartbeat/src/mybot/tools/__init__.py @@ -0,0 +1,7 @@ +"""Tools module for agent capabilities.""" + +from mybot.tools.base import BaseTool, tool +from mybot.tools.builtin_tools import bash, edit_file, read_file, write_file +from mybot.tools.registry import ToolRegistry + +__all__ = ["BaseTool", "tool", "ToolRegistry", "read_file", "write_file", "edit_file", "bash"] diff --git a/12-cron-heartbeat/src/mybot/tools/base.py b/12-cron-heartbeat/src/mybot/tools/base.py new file mode 100644 index 0000000..b2c7701 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/tools/base.py @@ -0,0 +1,63 @@ +"""Base tool interface and decorator.""" + +import asyncio +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Any, Callable + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class BaseTool(ABC): + """Abstract base class for all tools.""" + + name: str + description: str + parameters: dict[str, Any] # JSON Schema for function calling + + @abstractmethod + async def execute(self, session: "AgentSession", **kwargs: Any) -> str: + """Execute the tool.""" + + def get_tool_schema(self) -> dict[str, Any]: + """Get the tool/function schema for LiteLLM.""" + return { + "type": "function", + "function": { + "name": self.name, + "description": self.description, + "parameters": self.parameters, + }, + } + + +def tool(name: str, description: str, parameters: dict[str, Any]) -> Callable: + """Decorator to register a function as a tool.""" + + def decorator(func: Callable) -> "FunctionTool": + return FunctionTool(name, description, parameters, func) + + return decorator + + +class FunctionTool(BaseTool): + """A tool created from a function using the @tool decorator.""" + + def __init__( + self, + name: str, + description: str, + parameters: dict[str, Any], + func: Callable, + ): + self.name = name + self.description = description + self.parameters = parameters + self._func = func + + async def execute(self, session: "AgentSession", **kwargs: Any) -> str: + """Execute the underlying function.""" + result = self._func(session=session, **kwargs) + if asyncio.iscoroutine(result): + result = await result + return str(result) diff --git a/12-cron-heartbeat/src/mybot/tools/builtin_tools.py b/12-cron-heartbeat/src/mybot/tools/builtin_tools.py new file mode 100644 index 0000000..d327e14 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/tools/builtin_tools.py @@ -0,0 +1,133 @@ +"""Built-in tools for agent capabilities.""" + +import asyncio +from pathlib import Path +from typing import TYPE_CHECKING + +from mybot.tools.base import tool + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +# Filesystem tools + + +@tool( + name="read", + description="Read the contents of a text file", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to the file to read"}, + }, + "required": ["path"], + }, +) +async def read_file(path: str, session: "AgentSession") -> str: + """Read and return the contents of a file at the given path.""" + try: + return Path(path).read_text() + except FileNotFoundError: + return f"Error: File not found: {path}" + except PermissionError: + return f"Error: Permission denied reading: {path}" + except IsADirectoryError: + return f"Error: Path is a directory, not a file: {path}" + except Exception as e: + return f"Error reading file: {e}" + + +@tool( + name="write", + description="Write content to a file", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to the file to write"}, + "content": { + "type": "string", + "description": "Content to write to the file", + }, + }, + "required": ["path", "content"], + }, +) +async def write_file(path: str, content: str, session: "AgentSession") -> str: + """Write content to a file at the given path.""" + try: + Path(path).write_text(content) + return f"Successfully wrote to: {path}" + except PermissionError: + return f"Error: Permission denied writing to: {path}" + except IsADirectoryError: + return f"Error: Path is a directory, not a file: {path}" + except Exception as e: + return f"Error writing file: {e}" + + +@tool( + name="edit", + description="Edit a file by replacing a string with new content", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to the file to edit"}, + "old_text": {"type": "string", "description": "The text to replace"}, + "new_text": { + "type": "string", + "description": "The new text to replace with", + }, + }, + "required": ["path", "old_text", "new_text"], + }, +) +async def edit_file( + path: str, old_text: str, new_text: str, session: "AgentSession" +) -> str: + """Edit a file by replacing old_text with new_text.""" + try: + content = Path(path).read_text() + if old_text not in content: + return f"Error: '{old_text}' not found in {path}" + new_content = content.replace(old_text, new_text) + Path(path).write_text(new_content) + return f"Successfully edited {path}" + except FileNotFoundError: + return f"Error: File not found: {path}" + except PermissionError: + return f"Error: Permission denied editing: {path}" + except Exception as e: + return f"Error editing file: {e}" + + +# Shell tool + + +@tool( + name="bash", + description="Execute a bash shell command", + parameters={ + "type": "object", + "properties": { + "command": {"type": "string", "description": "The bash command to execute"}, + }, + "required": ["command"], + }, +) +async def bash(command: str, session: "AgentSession") -> str: + """Execute a bash command and return the output.""" + try: + process = await asyncio.create_subprocess_shell( + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + output = stdout.decode() if stdout else "" + error = stderr.decode() if stderr else "" + if output and error: + return f"{output}\n{error}" + return output or error or "Command completed with no output" + except Exception as e: + return f"Error executing command: {e}" diff --git a/12-cron-heartbeat/src/mybot/tools/registry.py b/12-cron-heartbeat/src/mybot/tools/registry.py new file mode 100644 index 0000000..2424099 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/tools/registry.py @@ -0,0 +1,56 @@ +"""Tool registry for managing available tools.""" + +from typing import TYPE_CHECKING, Any + +from mybot.tools.base import BaseTool +from mybot.tools.builtin_tools import bash, edit_file, read_file, write_file + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class ToolRegistry: + """Registry for all available tools.""" + + def __init__(self) -> None: + """Initialize an empty tool registry.""" + self._tools: dict[str, BaseTool] = {} + + def register(self, tool: BaseTool) -> None: + """Register a tool.""" + self._tools[tool.name] = tool + + def get(self, name: str) -> BaseTool | None: + """Get a tool by name.""" + return self._tools.get(name) + + def list_all(self) -> list[BaseTool]: + """List all registered tools.""" + return list(self._tools.values()) + + def get_tool_schemas(self) -> list[dict[str, Any]]: + """Get tool schemas for all registered tools.""" + return [tool.get_tool_schema() for tool in self._tools.values()] + + async def execute_tool( + self, name: str, session: "AgentSession", **kwargs: Any + ) -> str: + """Execute a tool by name.""" + tool = self.get(name) + if tool is None: + raise ValueError(f"Tool not found: {name}") + + return await tool.execute(session=session, **kwargs) + + @classmethod + def with_builtins(cls) -> "ToolRegistry": + """Create a ToolRegistry with builtin tools already registered.""" + + registry = cls() + + registry.register(read_file) + registry.register(write_file) + registry.register(edit_file) + registry.register(bash) + + return registry diff --git a/12-cron-heartbeat/src/mybot/tools/skill_tool.py b/12-cron-heartbeat/src/mybot/tools/skill_tool.py new file mode 100644 index 0000000..b4a37e0 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/tools/skill_tool.py @@ -0,0 +1,51 @@ +"""Skill tool factory for creating dynamic skill tool.""" + +from typing import TYPE_CHECKING + +from mybot.tools.base import tool + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + from mybot.core.skill_loader import SkillLoader + + +def create_skill_tool(skill_loader: "SkillLoader"): + """Factory function to create skill tool with dynamic schema.""" + skill_metadata = skill_loader.discover_skills() + + if not skill_metadata: + return None + + # Build XML description of available skills + skills_xml = "\n" + for meta in skill_metadata: + skills_xml += f' {meta.description}\n' + skills_xml += "" + + # Build enum of skill IDs + skill_enum = [meta.id for meta in skill_metadata] + + @tool( + name="skill", + description=f"Load and invoke a specialized skill. {skills_xml}", + parameters={ + "type": "object", + "properties": { + "skill_name": { + "type": "string", + "enum": skill_enum, + "description": "The name of the skill to load", + } + }, + "required": ["skill_name"], + }, + ) + async def skill_tool(skill_name: str, session: "AgentSession") -> str: + """Load and return skill content.""" + try: + skill_def = skill_loader.load_skill(skill_name) + return skill_def.content + except Exception: + return f"Error: Skill '{skill_name}' not found. It may have been removed or is unavailable." + + return skill_tool diff --git a/12-cron-heartbeat/src/mybot/tools/webread_tool.py b/12-cron-heartbeat/src/mybot/tools/webread_tool.py new file mode 100644 index 0000000..7160b2c --- /dev/null +++ b/12-cron-heartbeat/src/mybot/tools/webread_tool.py @@ -0,0 +1,47 @@ +"""Webread tool factory.""" + +from typing import TYPE_CHECKING + +from mybot.tools.base import BaseTool, tool +from mybot.provider.web_read import WebReadProvider + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + from mybot.core.context import SharedContext + + +def create_webread_tool(context: "SharedContext") -> BaseTool | None: + """Factory to create webread tool with injected context.""" + if not context.config.webread: + return None + + provider = WebReadProvider.from_config(context.config) + + @tool( + name="webread", + description=( + "Read and extract content from a web page. " + "Returns the page content as markdown." + ), + parameters={ + "type": "object", + "properties": { + "url": { + "type": "string", + "description": "The URL to read", + } + }, + "required": ["url"], + }, + ) + async def webread(url: str, session: "AgentSession") -> str: + """Read a web page and return markdown content.""" + + result = await provider.read(url) + + if result.error: + return f"Error reading {url}: {result.error}" + + return f"**{result.title}**\n\n{result.content}" + + return webread diff --git a/12-cron-heartbeat/src/mybot/tools/websearch_tool.py b/12-cron-heartbeat/src/mybot/tools/websearch_tool.py new file mode 100644 index 0000000..bc3a319 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/tools/websearch_tool.py @@ -0,0 +1,50 @@ +"""Websearch tool factory.""" + +from typing import TYPE_CHECKING + +from mybot.tools.base import BaseTool, tool +from mybot.provider.web_search import WebSearchProvider + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + from mybot.core.context import SharedContext + + +def create_websearch_tool(context: "SharedContext") -> BaseTool | None: + """Factory to create websearch tool with injected context.""" + if not context.config.websearch: + return None + + provider = WebSearchProvider.from_config(context.config) + + @tool( + name="websearch", + description=( + "Search the web for information. " + "Returns a list of results with titles, URLs, and snippets." + ), + parameters={ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "The search query", + } + }, + "required": ["query"], + }, + ) + async def websearch(query: str, session: "AgentSession") -> str: + """Search the web and return formatted results.""" + + results = await provider.search(query) + + if not results: + return "No results found." + + output = [] + for i, r in enumerate(results, 1): + output.append(f"{i}. **{r.title}**\n {r.url}\n {r.snippet}") + return "\n\n".join(output) + + return websearch diff --git a/12-cron-heartbeat/src/mybot/utils/__init__.py b/12-cron-heartbeat/src/mybot/utils/__init__.py new file mode 100644 index 0000000..136c49a --- /dev/null +++ b/12-cron-heartbeat/src/mybot/utils/__init__.py @@ -0,0 +1,17 @@ +"""Utilities package.""" + +from mybot.utils.def_loader import ( + DefNotFoundError, + InvalidDefError, + discover_definitions, + parse_definition, +) +from mybot.utils.logging import setup_logging + +__all__ = [ + "DefNotFoundError", + "InvalidDefError", + "discover_definitions", + "parse_definition", + "setup_logging", +] diff --git a/12-cron-heartbeat/src/mybot/utils/config.py b/12-cron-heartbeat/src/mybot/utils/config.py new file mode 100644 index 0000000..44c6a99 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/utils/config.py @@ -0,0 +1,244 @@ +"""Configuration management with hot reload support.""" + +import logging +from pathlib import Path +from typing import Any, Literal + +import yaml +from pydantic import BaseModel, Field, field_validator, model_validator +from watchdog.events import FileSystemEventHandler +from watchdog.observers import Observer + + +class LLMConfig(BaseModel): + """LLM provider configuration.""" + + provider: str + model: str + api_key: str + api_base: str | None = None + temperature: float = Field(default=0.7, ge=0.0, le=2.0) + max_tokens: int = Field(default=2048, gt=0) + + @field_validator("api_base") + @classmethod + def api_base_must_be_url(cls, v: str | None) -> str | None: + if v is not None and not v.startswith(("http://", "https://")): + raise ValueError("api_base must be a valid URL") + return v + + +class TelegramConfig(BaseModel): + """Telegram platform configuration.""" + + enabled: bool = True + bot_token: str + allowed_user_ids: list[str] = Field(default_factory=list) + + +class DiscordConfig(BaseModel): + """Discord platform configuration.""" + + enabled: bool = True + bot_token: str + channel_id: str | None = None + allowed_user_ids: list[str] = Field(default_factory=list) + + +class BraveWebSearchConfig(BaseModel): + """Configuration for web search provider.""" + + provider: Literal["brave"] = "brave" + api_key: str + + +class Crawl4AIWebReadConfig(BaseModel): + """Configuration for web read provider.""" + + provider: Literal["crawl4ai"] = "crawl4ai" + + +class SourceSessionConfig(BaseModel): + """Session affinity configuration for a source.""" + + session_id: str + + +class ChannelConfig(BaseModel): + """Channel configuration.""" + + enabled: bool = False + telegram: TelegramConfig | None = None + discord: DiscordConfig | None = None + + +class ApiConfig(BaseModel): + """HTTP API configuration.""" + + host: str = "127.0.0.1" + port: int = Field(default=8000, gt=0, lt=65536) + + +class Config(BaseModel): + """Main configuration with hot reload support.""" + + workspace: Path + llm: LLMConfig + default_agent: str + agents_path: Path = Field(default=Path("agents")) + skills_path: Path = Field(default=Path("skills")) + crons_path: Path = Field(default=Path("crons")) + logging_path: Path = Field(default=Path(".logs")) + history_path: Path = Field(default=Path(".history")) + event_path: Path = Field(default=Path(".event")) + websearch: BraveWebSearchConfig | None = None + webread: Crawl4AIWebReadConfig | None = None + channels: ChannelConfig = Field(default_factory=ChannelConfig) + api: ApiConfig = Field(default_factory=ApiConfig) + sources: dict[str, SourceSessionConfig] = Field(default_factory=dict) + routing: dict = Field(default_factory=lambda: {"bindings": []}) + default_delivery_source: str | None = None + + @model_validator(mode="after") + def resolve_paths(self) -> "Config": + """Resolve relative paths to absolute using workspace.""" + for field_name in ( + "agents_path", + "skills_path", + "crons_path", + "logging_path", + "history_path", + "event_path", + ): + path = getattr(self, field_name) + if not path.is_absolute(): + setattr(self, field_name, self.workspace / path) + return self + + @classmethod + def load(cls, workspace_dir: Path) -> "Config": + """Load configuration from workspace directory.""" + config_data = cls._load_merged_configs(workspace_dir) + config_data["workspace"] = workspace_dir + return cls.model_validate(config_data) + + @classmethod + def _load_merged_configs(cls, workspace_dir: Path) -> dict[str, Any]: + """Load and merge user and runtime config files.""" + config_data: dict[str, Any] = {} + + user_config = workspace_dir / "config.user.yaml" + runtime_config = workspace_dir / "config.runtime.yaml" + + if user_config.exists(): + with open(user_config) as f: + config_data = cls._deep_merge(config_data, yaml.safe_load(f) or {}) + + if runtime_config.exists(): + with open(runtime_config) as f: + config_data = cls._deep_merge(config_data, yaml.safe_load(f) or {}) + + return config_data + + @staticmethod + def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + """Deep merge override dict into base dict.""" + result = base.copy() + + for key, value in override.items(): + if ( + key in result + and isinstance(result[key], dict) + and isinstance(value, dict) + ): + result[key] = Config._deep_merge(result[key], value) + else: + result[key] = value + + return result + + def _set_nested(self, obj: dict, key: str, value: Any) -> None: + """Set a nested value in a dict using dot notation.""" + keys = key.split(".") + for k in keys[:-1]: + if k not in obj or not isinstance(obj[k], dict): + obj[k] = {} + obj = obj[k] + obj[keys[-1]] = value + + def _set_config_value(self, config_path: Path, key: str, value: Any) -> None: + """Update a config value in a YAML file.""" + # Load existing or start fresh + if config_path.exists(): + with open(config_path) as f: + data = yaml.safe_load(f) or {} + else: + data = {} + + if isinstance(value, BaseModel): + value = value.model_dump() + + # Update the key (supports nested via dot notation) + self._set_nested(data, key, value) + + # Write back + with open(config_path, "w") as f: + yaml.dump(data, f) + + def set_user(self, key: str, value: Any) -> None: + """Update a config value in config.user.yaml.""" + self._set_config_value(self.workspace / "config.user.yaml", key, value) + + def set_runtime(self, key: str, value: Any) -> None: + """Update a runtime value in config.runtime.yaml.""" + self._set_config_value(self.workspace / "config.runtime.yaml", key, value) + + def reload(self) -> bool: + """Re-read config.user.yaml and merge with runtime.""" + try: + config_data = self._load_merged_configs(self.workspace) + config_data["workspace"] = self.workspace + + # Create new instance and copy values + new_config = Config.model_validate(config_data) + + # Update all fields from new config + for field_name in Config.model_fields: + setattr(self, field_name, getattr(new_config, field_name)) + + return True + except Exception as e: + logging.debug("Config reload failed: %s", e) + return False + + +class ConfigHandler(FileSystemEventHandler): + """Handles config file modification events.""" + + def __init__(self, config: Config): + self._config = config + + def on_modified(self, event): + """Reload config when config.user.yaml changes.""" + if not event.is_directory and event.src_path.endswith("config.user.yaml"): + self._config.reload() + + +class ConfigReloader: + """Manages watchdog observer for config hot reload.""" + + def __init__(self, config: Config): + self._config = config + self._observer = Observer() + + def start(self) -> None: + """Start watching config file for changes.""" + handler = ConfigHandler(self._config) + self._observer.schedule(handler, str(self._config.workspace), recursive=False) + self._observer.start() + + def stop(self) -> None: + """Stop watching.""" + self._observer.stop() + self._observer.join() + del self._observer diff --git a/12-cron-heartbeat/src/mybot/utils/def_loader.py b/12-cron-heartbeat/src/mybot/utils/def_loader.py new file mode 100644 index 0000000..6175837 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/utils/def_loader.py @@ -0,0 +1,105 @@ +"""Shared utilities for loading definition files (agents, skills, crons).""" + +import logging +from pathlib import Path +from typing import Any, Callable, TypeVar + +import yaml + +T = TypeVar("T") +logger = logging.getLogger(__name__) + + +class DefNotFoundError(Exception): + """Definition folder or file doesn't exist.""" + + def __init__(self, kind: str, def_id: str): + super().__init__(f"{kind.capitalize()} not found: {def_id}") + self.kind = kind + self.def_id = def_id + + +class InvalidDefError(Exception): + """Definition file is malformed.""" + + def __init__(self, kind: str, def_id: str, reason: str): + super().__init__(f"Invalid {kind} '{def_id}': {reason}") + self.kind = kind + self.def_id = def_id + self.reason = reason + + +def parse_definition( + content: str, + def_id: str, + parse_fn: Callable[[str, dict[str, Any], str], T], +) -> T: + """Parse YAML frontmatter + markdown body with type conversion.""" + # Find frontmatter delimiters + if not content.startswith("---\n"): + body = content + return parse_fn(def_id, {}, body) + + end_delimiter = content.find("\n---\n", 4) + if end_delimiter == -1: + body = content + return parse_fn(def_id, {}, body) + + frontmatter_text = content[4:end_delimiter] + body = content[end_delimiter + 5 :] + + raw_dict = yaml.safe_load(frontmatter_text) or {} + return parse_fn(def_id, raw_dict, body) + + +def discover_definitions( + path: Path, + filename: str, + parse_fn: Callable[[str, dict[str, Any], str], T | None], +) -> list[T]: + """Scan directory for definition files.""" + if not path.exists(): + logger.warning(f"Definitions directory not found: {path}") + return [] + + results = [] + for def_dir in path.iterdir(): + if not def_dir.is_dir(): + continue + + def_file = def_dir / filename + if not def_file.exists(): + logger.warning(f"No {filename} found in {def_dir.name}") + continue + + try: + content = def_file.read_text() + result = parse_definition(content, def_dir.name, parse_fn) + if result is not None: + results.append(result) + except Exception as e: + logger.warning(f"Failed to parse {def_dir.name}: {e}") + continue + + return results + + +def write_definition( + def_id: str, + frontmatter: dict[str, Any], + body: str, + base_path: Path, + filename: str, +) -> Path: + """Write a definition file with YAML frontmatter and markdown body.""" + def_dir = base_path / def_id + def_dir.mkdir(parents=True, exist_ok=True) + + # Build file content with YAML frontmatter + yaml_content = yaml.dump(frontmatter, default_flow_style=False, sort_keys=False) + content = f"---\n{yaml_content}---\n\n{body.strip()}\n" + + def_file = def_dir / filename + def_file.write_text(content) + + return def_file diff --git a/12-cron-heartbeat/src/mybot/utils/logging.py b/12-cron-heartbeat/src/mybot/utils/logging.py new file mode 100644 index 0000000..bf1bb71 --- /dev/null +++ b/12-cron-heartbeat/src/mybot/utils/logging.py @@ -0,0 +1,34 @@ +"""Logging configuration for mybot.""" + +import logging +import sys + +from mybot.utils.config import Config +from logging.handlers import RotatingFileHandler + +def setup_logging(config: Config, console_output: bool = False) -> None: + """Set up logging for pickle-bot.""" + format_str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + formatter = logging.Formatter(format_str) + + # Console format is simpler (no timestamp) + console_format = "%(levelname)s - %(name)s - %(message)s" + console_formatter = logging.Formatter(console_format) + + root_logger = logging.getLogger("mybot") + root_logger.setLevel(logging.DEBUG) + + config.logging_path.mkdir(parents=True, exist_ok=True) + file_handler = RotatingFileHandler( + config.logging_path / "mybot.log", maxBytes=256 * 1024 * 128, backupCount=3 + ) + file_handler.setFormatter(formatter) + file_handler.setLevel(logging.DEBUG) + root_logger.addHandler(file_handler) + + # Optionally log to console (for server mode) + if console_output: + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setFormatter(console_formatter) + console_handler.setLevel(logging.INFO) + root_logger.addHandler(console_handler) diff --git a/PLAN.md b/PLAN.md index 3f12f50..883be8c 100644 --- a/PLAN.md +++ b/PLAN.md @@ -562,7 +562,34 @@ class ContextGuard: --- -### Step 13: Post Message Back - Agent-Initiated Communication +### Step 13: Multi-Layer Prompts + +**Problem:** Single AGENT.md isn't flexible enough + +**What to Build:** +- SOUL.md (personality) +- MEMORY.md (persistent knowledge) +- Prompt composition +- Layer loading + +**Pickle-bot References:** +- `core/agent_def.py` - Multi-file loading + +**Implementation Notes:** +- AGENT.md = base prompt (capabilities, tools) +- SOUL.md = personality layer (tone, style) +- MEMORY.md = knowledge layer (user info) +- Compose all layers at load time +- Show in AGENT.md what layers exist + +**Alternative Approaches:** +1. Single AGENT.md only (simpler, less organized) +2. Use database for memory (more complex, more structured) +3. Use config file instead of markdown (less readable) + +--- + +### Step 14: Post Message Back - Agent-Initiated Communication **Problem:** Agent can't initiate outbound messages @@ -588,7 +615,7 @@ class ContextGuard: --- -### Step 14: Agent Dispatch - Multi-Agent Collaboration +### Step 15: Agent Dispatch - Multi-Agent Collaboration **Problem:** Need specialized agents working together @@ -617,33 +644,6 @@ class ContextGuard: --- -### Step 15: Multi-Layer Prompts - -**Problem:** Single AGENT.md isn't flexible enough - -**What to Build:** -- SOUL.md (personality) -- MEMORY.md (persistent knowledge) -- Prompt composition -- Layer loading - -**Pickle-bot References:** -- `core/agent_def.py` - Multi-file loading - -**Implementation Notes:** -- AGENT.md = base prompt (capabilities, tools) -- SOUL.md = personality layer (tone, style) -- MEMORY.md = knowledge layer (user info) -- Compose all layers at load time -- Show in AGENT.md what layers exist - -**Alternative Approaches:** -1. Single AGENT.md only (simpler, less organized) -2. Use database for memory (more complex, more structured) -3. Use config file instead of markdown (less readable) - ---- - ## Phase 4: Production & Scale ### Step 16: Concurrency Control @@ -724,9 +724,9 @@ build-your-own-openclaw/ ├── 10-websocket/ ├── 11-multi-agent-routing/ ├── 12-cron-heartbeat/ -├── 13-post-message-back/ -├── 14-agent-dispatch/ -├── 15-multi-layer-prompts/ +├── 13-multi-layer-prompts/ +├── 14-post-message-back/ +├── 15-agent-dispatch/ ├── 16-concurrency-control/ └── 17-memory/ ├── README.md diff --git a/README.md b/README.md index be08272..6f6f136 100644 --- a/README.md +++ b/README.md @@ -42,9 +42,9 @@ Add scheduled tasks, agent collaboration, and intelligent routing. - **11-multi-agent-routing** - Multiple agent & Right agent for right job - **12-cron-heartbeat** - Autonomous scheduled tasks -- **13-post-message-back** - Agent-initiated communication -- **14-agent-dispatch** - Agent collaboration -- **15-multi-layer-prompts** - Sophisticated agent configuration +- **13-multi-layer-prompts** - Responsive system prompt +- **14-post-message-back** - Agent-initiated communication +- **15-agent-dispatch** - Agent collaboration ### Phase 4: Production & Scale (Steps 17-18) Production features for reliability and long-term memory.