17-memory

This commit is contained in:
zane
2026-03-11 20:33:33 -04:00
parent 519f437b7d
commit a11742e5a9
61 changed files with 4852 additions and 990 deletions
+95
View File
@@ -0,0 +1,95 @@
# Step 17: Memory - Long-Term Knowledge System
A specialized memory agent (Cookie) that manages persistent knowledge via agent dispatch.
## 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 Built
### Architecture
```
User ↔ Pickle (Main Agent)
↓ (dispatch)
Cookie (Memory Agent)
↓ (tools: read, write, bash)
memories/
├── topics/ (preferences, identity)
├── projects/ (project-specific context)
└── daily-notes/ (YYYY-MM-DD.md)
```
### Key Components
- **Memory agent**: Specialized agent for memory management
## Key Changes
### `default_workspace/agents/cookie/AGENT.md` - Memory agent definition
Already exists from picklebot. Defines Cookie's role:
- Store memories using `write` tool
- Retrieve memories using `read` tool
- Organize by topics (timeless), projects, and daily notes
- Only accessible via dispatch from Pickle
## How to Use
The memory system is now configured. Pickle can dispatch to Cookie:
```
User: Remember that I prefer TypeScript for new projects
Pickle: (dispatches to Cookie)
Cookie: (stores preference in memories/topics/preferences.md)
Pickle: Got it! I've saved that preference.
```
Later:
```
User: What language should I use for my new API?
Pickle: (dispatches to Cookie to retrieve preferences)
Cookie: (reads memories/topics/preferences.md)
Pickle: Based on your preferences, I recommend TypeScript.
```
## Discussion: Memory System Approaches
This is **one approach** to implementing long-term memory. There are many alternatives:
1. **Specialized Agent** (this implementation)
2. **Direct Tools in Main Agent**
3. **Skill Based Approach, using command line tool like grep**
3. **Vector Database**
## How to Run
```bash
cd 17-memory
uv run my-bot chat
# Memory path is now available in config
# Cookie agent is ready for dispatch
# Try: "Ask Cookie to remember something for me"
```
## What's Next
This completes the tutorial! You now have a production-ready agent with:
- ✅ Event-driven architecture
- ✅ Multi-platform support
- ✅ Multi-agent collaboration
- ✅ Scheduled tasks
- ✅ Concurrency control
- ✅ Long-term memory
Next steps: Deploy, extend, and customize for your specific use case!
+32
View File
@@ -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"
+7
View File
@@ -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"]
+57
View File
@@ -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
@@ -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")
@@ -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")
+5
View File
@@ -0,0 +1,5 @@
"""CLI interface for my-bot."""
from mybot.cli.main import app
__all__ = ["app"]
+125
View File
@@ -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())
+80
View File
@@ -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()
+25
View File
@@ -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")
+20
View File
@@ -0,0 +1,20 @@
"""Core agent functionality."""
from .agent import Agent, AgentSession
from .agent_loader import (
AgentLoader,
AgentDef,
)
from .context import SharedContext
from .history import HistoryMessage, HistorySession, HistoryStore
__all__ = [
"Agent",
"AgentSession",
"AgentDef",
"AgentLoader",
"SharedContext",
"HistoryStore",
"HistoryMessage",
"HistorySession",
]
+252
View File
@@ -0,0 +1,252 @@
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 mybot.tools.post_message_tool import create_post_message_tool
from mybot.tools.subagent_tool import create_subagent_dispatch_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, include_post_message: bool) -> 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)
if include_post_message:
post_tool = create_post_message_tool(self.context)
if post_tool:
registry.register(post_tool)
# Register subagent dispatch tool
subagent_tool = create_subagent_dispatch_tool(
self.agent_def.id, self.context
)
if subagent_tool:
registry.register(subagent_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())
include_post_message = source.is_cron
tools = self._build_tools(include_post_message)
# 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()
include_post_message = source.is_cron
# 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(include_post_message)
# 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
+98
View File
@@ -0,0 +1,98 @@
"""Agent definition loader."""
from typing import Any
from pydantic import BaseModel, Field, 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
soul_md: str = "" # Personality layer (optional)
llm: LLMConfig
allow_skills: bool = False
max_concurrency: int = Field(default=1, ge=1)
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)
# Load SOUL.md if exists
soul_md = self._load_soul_md(def_id)
try:
return AgentDef(
id=def_id,
name=frontmatter["name"], # type: ignore[misc]
description=frontmatter.get("description", ""),
agent_md=body.strip(),
soul_md=soul_md,
llm=merged_llm,
allow_skills=frontmatter.get("allow_skills", False),
max_concurrency=frontmatter.get("max_concurrency", 1),
)
except ValidationError as e:
raise InvalidDefError("agent", def_id, str(e))
def _load_soul_md(self, agent_id: str) -> str:
"""Load SOUL.md file for an agent if it exists."""
soul_path = self.config.agents_path / agent_id / "SOUL.md"
if soul_path.exists():
return soul_path.read_text().strip()
return ""
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)
@@ -0,0 +1,6 @@
"""Slash commands module."""
from mybot.core.commands.base import Command
from mybot.core.commands.registry import CommandRegistry
__all__ = ["Command", "CommandRegistry"]
+20
View File
@@ -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
@@ -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 <source_pattern> <agent_id>`\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)
@@ -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
+51
View File
@@ -0,0 +1,51 @@
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.prompt_builder import PromptBuilder
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
prompt_builder: PromptBuilder
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)
self.prompt_builder = PromptBuilder(self)
if channels is not None:
self.channels = channels
else:
self.channels = Channel.from_config(config)
self.eventbus = EventBus(self)
self.websocket_worker = None
+158
View File
@@ -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
+114
View File
@@ -0,0 +1,114 @@
"""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
self.config.crons_path.mkdir(parents=True, exist_ok=True)
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
+145
View File
@@ -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}")
+211
View File
@@ -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)
+230
View File
@@ -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
@@ -0,0 +1,91 @@
"""Prompt builder that assembles system prompt from layers."""
from datetime import datetime
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from mybot.core.context import SharedContext
from mybot.core.events import EventSource
from mybot.core.session_state import SessionState
class PromptBuilder:
"""Assembles system prompt from layered sources."""
def __init__(self, context: "SharedContext"):
self.context = context
def build(self, state: "SessionState") -> str:
"""Build the full system prompt from layers."""
layers = []
# Layer 1: Identity
layers.append(state.agent.agent_def.agent_md)
# Layer 2: Soul (optional)
if state.agent.agent_def.soul_md:
layers.append(f"## Personality\n\n{state.agent.agent_def.soul_md}")
# Layer 3: Bootstrap context
bootstrap = self._load_bootstrap_context()
if bootstrap:
layers.append(bootstrap)
# Layer 4: Runtime context
layers.append(
self._build_runtime_context(
state.agent.agent_def.id,
datetime.now(),
)
)
# Layer 5: Channel hint
layers.append(self._build_channel_hint(state.source))
return "\n\n".join(layers)
def _load_bootstrap_context(self) -> str:
"""Load BOOTSTRAP.md + AGENTS.md + cron list."""
parts = []
bootstrap_path = self.context.config.workspace / "BOOTSTRAP.md"
if bootstrap_path.exists():
parts.append(bootstrap_path.read_text().strip())
agents_path = self.context.config.workspace / "AGENTS.md"
if agents_path.exists():
parts.append(agents_path.read_text().strip())
# Dynamic cron list
cron_list = self._format_cron_list()
if cron_list:
parts.append(cron_list)
return "\n\n".join(parts)
def _format_cron_list(self) -> str:
"""Format crons as markdown list."""
crons = self.context.cron_loader.discover_crons()
if not crons:
return ""
lines = ["## Scheduled Tasks\n"]
for cron in crons:
lines.append(f"- **{cron.name}**: {cron.description}")
return "\n".join(lines)
def _build_runtime_context(self, agent_id: str, timestamp: datetime) -> str:
"""Build runtime info section."""
return f"## Runtime\n\nAgent: {agent_id}\nTime: {timestamp.isoformat()}"
def _build_channel_hint(self, source: "EventSource") -> str:
"""Build platform hint."""
if source.is_cron:
return "You are running as a background cron job. Your response will not be sent to user directly."
if source.is_agent:
return "You are running as a dispatched subagent. Your response will be sent to main agent."
elif source.is_platform:
return f"You are responding via {source.platform_name}."
else:
raise ValueError(f"Unknown source type: {source}")
+113
View File
@@ -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)
)
+37
View File
@@ -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.shared_context.prompt_builder.build(self)
messages: list[Message] = [{"role": "system", "content": system_prompt}]
messages.extend(self.messages)
return messages
+69
View File
@@ -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)
@@ -0,0 +1,5 @@
"""LLM provider abstraction."""
from .base import LLMProvider, LLMToolCall
__all__ = ["LLMProvider", "LLMToolCall"]
+86
View File
@@ -0,0 +1,86 @@
"""Base LLM provider abstraction."""
from dataclasses import dataclass
from typing import Any, Optional, cast
from litellm import acompletion, Choices, TYPE_CHECKING
from litellm.types.completion import ChatCompletionMessageParam as Message
if TYPE_CHECKING:
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 [])
],
)
@@ -0,0 +1,5 @@
"""Web read provider module."""
from .base import ReadResult, WebReadProvider
__all__ = ["ReadResult", "WebReadProvider"]
@@ -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}")
@@ -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),
)
@@ -0,0 +1,5 @@
"""Web search provider module."""
from .base import SearchResult, WebSearchProvider
__all__ = ["SearchResult", "WebSearchProvider"]
@@ -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}"
)
@@ -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
+18
View File
@@ -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",
]
+178
View File
@@ -0,0 +1,178 @@
"""Agent worker for executing agent jobs."""
import asyncio
import logging
from dataclasses import replace
from typing import TYPE_CHECKING, 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
if TYPE_CHECKING:
from mybot.core.context import SharedContext
from mybot.core.agent_loader import AgentDef
# Maximum number of retry attempts for failed sessions
MAX_RETRIES = 3
logger = logging.getLogger(__name__)
ProcessableEvent = Union[InboundEvent, DispatchEvent]
class AgentWorker(SubscriberWorker):
"""Dispatches events to session executors with per-agent concurrency control.
Auto-subscribes to:
- InboundEvent (from platforms, cron, retries)
- DispatchEvent (from subagent calls)
"""
CLEANUP_THRESHOLD = 5
def __init__(self, context: "SharedContext"):
super().__init__(context)
self._semaphores: dict[str, asyncio.Semaphore] = {}
# 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: ProcessableEvent) -> 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_id=agent_id,
content="",
error=str(e),
)
asyncio.create_task(self.exec_session(event, agent_def))
async def exec_session(
self, event: ProcessableEvent, agent_def: "AgentDef"
) -> None:
sem = self._get_or_create_semaphore(agent_def)
session_id = event.session_id
async with sem:
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(event.source, session_id=session_id)
else:
session = agent.new_session(event.source)
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, content=result, agent_id=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,
agent_id=agent_def.id,
content=response,
)
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,
content="",
agent_id=agent_def.id,
error=str(e),
)
self._maybe_cleanup_semaphores(agent_def)
async def _emit_response(
self,
event: ProcessableEvent,
content: str,
agent_id: str,
error: str | None = None,
) -> None:
"""Emit response event with content."""
if isinstance(event, DispatchEvent):
result_event: DispatchResultEvent | OutboundEvent = 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)
def _get_or_create_semaphore(self, agent_def: "AgentDef") -> asyncio.Semaphore:
"""Get existing or create new semaphore for agent."""
if agent_def.id not in self._semaphores:
self._semaphores[agent_def.id] = asyncio.Semaphore(
agent_def.max_concurrency
)
logger.debug(
f"Created semaphore for {agent_def.id} with value {agent_def.max_concurrency}"
)
return self._semaphores[agent_def.id]
def _maybe_cleanup_semaphores(self, agent_def: "AgentDef") -> None:
"""Remove semaphores for certain agents."""
if agent_def.id not in self._semaphores:
return
if not self._semaphores[agent_def.id]._waiters:
del self._semaphores[agent_def.id]
+41
View File
@@ -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
@@ -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
+84
View File
@@ -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}")
@@ -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
+123
View File
@@ -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()
@@ -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)
+60
View File
@@ -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
+7
View File
@@ -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"]
+63
View File
@@ -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)
+133
View File
@@ -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}"
@@ -0,0 +1,55 @@
"""Post message tool factory for agent-initiated messaging."""
import time
from typing import TYPE_CHECKING
from mybot.core.events import AgentEventSource, OutboundEvent
from mybot.tools.base import BaseTool, tool
if TYPE_CHECKING:
from mybot.core.agent import AgentSession
from mybot.core.context import SharedContext
def create_post_message_tool(context: "SharedContext") -> BaseTool | None:
"""Factory to create post_message tool."""
config = context.config
# Return None if channels not enabled or no channels configured
if not config.channels.enabled:
return None
# Check if we have any channels configured
if not context.channels:
return None
@tool(
name="post_message",
description="Send a message to the user via the default messaging platform. Use this to proactively notify the user about completed tasks, cron results, or important updates.",
parameters={
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "The message content to send to the user",
}
},
"required": ["content"],
},
)
async def post_message(content: str, session: "AgentSession") -> str:
"""Send a message to the default user on the default platform."""
try:
# Publish OUTBOUND event for the DeliveryWorker to handle
event = OutboundEvent(
session_id=session.session_id,
source=AgentEventSource(agent_id=session.agent.agent_def.id),
content=content,
timestamp=time.time(),
)
await context.eventbus.publish(event)
return "Message queued for delivery"
except Exception as e:
return f"Failed to send message: {e}"
return post_message
+56
View File
@@ -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
+51
View File
@@ -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 = "<skills>\n"
for meta in skill_metadata:
skills_xml += f' <skill name="{meta.name}">{meta.description}</skill>\n'
skills_xml += "</skills>"
# 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
+122
View File
@@ -0,0 +1,122 @@
"""Subagent dispatch tool factory for creating dynamic dispatch tool."""
import asyncio
import json
import time
from typing import TYPE_CHECKING
from mybot.core.events import (
AgentEventSource,
DispatchEvent,
DispatchResultEvent,
)
from mybot.tools.base import BaseTool, tool
from mybot.utils.def_loader import DefNotFoundError
if TYPE_CHECKING:
from mybot.core.agent import AgentSession
from mybot.core.context import SharedContext
def create_subagent_dispatch_tool(
current_agent_id: str,
context: "SharedContext",
) -> BaseTool | None:
"""Factory to create subagent dispatch tool with dynamic schema."""
# Discover available agents, exclude current
shared_context = context
available_agents = shared_context.agent_loader.discover_agents()
dispatchable_agents = [a for a in available_agents if a.id != current_agent_id]
if not dispatchable_agents:
return None
# Build description listing available agents
agents_desc = "<available_agents>\n"
for agent_def in dispatchable_agents:
agents_desc += f' <agent id="{agent_def.id}">{agent_def.description}</agent>\n'
agents_desc += "</available_agents>"
dispatchable_ids = [a.id for a in dispatchable_agents]
@tool(
name="subagent_dispatch",
description=f"Dispatch a task to a specialized subagent.\n{agents_desc}",
parameters={
"type": "object",
"properties": {
"agent_id": {
"type": "string",
"enum": dispatchable_ids,
"description": "ID of the agent to dispatch to",
},
"task": {
"type": "string",
"description": "The task for the subagent to perform",
},
"context": {
"type": "string",
"description": "Optional context information for the subagent",
},
},
"required": ["agent_id", "task"],
},
)
async def subagent_dispatch(
agent_id: str, task: str, session: "AgentSession", context: str = ""
) -> str:
"""Dispatch task to subagent, return result + session_id."""
# Verify agent exists and create session
from mybot.core.agent import Agent
try:
agent_def = shared_context.agent_loader.load(agent_id)
except DefNotFoundError:
raise ValueError(f"Agent '{agent_id}' not found")
agent = Agent(agent_def, shared_context)
agent_source = AgentEventSource(agent_id=current_agent_id)
agent_session = agent.new_session(agent_source)
session_id = agent_session.session_id
user_message = task
if context:
user_message = f"{task}\n\nContext:\n{context}"
loop = asyncio.get_running_loop()
result_future: asyncio.Future[str] = loop.create_future()
# Create temp handler that filters by session_id
async def handle_result(event: DispatchResultEvent) -> None:
if event.session_id == session_id:
if not result_future.done():
if event.error:
result_future.set_exception(Exception(event.error))
else:
result_future.set_result(event.content)
# Subscribe to DispatchResultEvent events
shared_context.eventbus.subscribe(DispatchResultEvent, handle_result)
try:
# Publish DISPATCH event
event = DispatchEvent(
session_id=session_id,
source=AgentEventSource(agent_id=current_agent_id),
content=user_message,
timestamp=time.time(),
parent_session_id=session.session_id,
)
await shared_context.eventbus.publish(event)
# Wait for result
response = await result_future
finally:
# Always unsubscribe
shared_context.eventbus.unsubscribe(handle_result)
result = {"result": response, "session_id": session_id}
return json.dumps(result)
return subagent_dispatch
+47
View File
@@ -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
@@ -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
+17
View File
@@ -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",
]
+246
View File
@@ -0,0 +1,246 @@
"""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"))
memories_path: Path = Field(default=Path("memories"))
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",
"memories_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
+105
View File
@@ -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
+34
View File
@@ -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)
-758
View File
@@ -1,758 +0,0 @@
# Implementation Plan - Build Your Own OpenClaw
This document provides detailed implementation guidance for each tutorial step, including specific pickle-bot code references and implementation notes.
## Implementation Principles
### **CRITICAL: Reuse pickle-bot Code**
- **Copy from pickle-bot** whenever possible - don't reinvent the wheel
- **Simplify only when needed** - remove production complexity, keep core logic
- **Reference paths** are relative to `../pickle-bot/src/picklebot/`
- **Keep it working** - each step must be runnable
### **Implementation Priority**
Follow this order when implementing each step:
1.**Match pickle-bot code first**
- Check `../pickle-bot/src/picklebot/` for existing implementation
- Copy classes, functions, patterns from pickle-bot
- Simplify if needed, but keep core logic
- Reference the source file in comments
2.**Match previous step second**
- Start from previous step's `src/` folder
- Make incremental changes only
- Preserve structure and patterns
- Show clear progression
3.**Write new code last**
- Only if no reference exists in pickle-bot
- Only if can't build on previous step
- Document why new code was needed
- Keep it minimal and focused
### **Incremental Approach**
- Each step starts from previous step's codebase
- Add ONE major feature per step
- Show the evolution, not just the result
- **No test files** - one session per step for implementation
### **CRITICAL: Follow the Workflow**
See [WORKFLOW.md](./WORKFLOW.md) for the step-by-step implementation process.
This ensures consistency with picklebot patterns across all steps.
---
## Phase 1: Capable Single Agent
### Step 00: Chat Loop - Your First Agent
**Problem:** Need basic agent that can have a conversation
**What to Build:**
- Minimal AgentSession class
- Basic config (API key, model name)
- AGENT.md loader (just system prompt)
- CLI chat loop with typer
**Pickle-bot References:**
- `core/agent.py` - AgentSession class (lines 1-80 for minimal version)
- `core/config.py` - Config loading
- `cli/main.py` - CLI structure
- `core/agent_def.py` - AGENT.md parsing
**Key Code to Copy:**
```python
# From: core/agent.py
class AgentSession:
def __init__(self, agent_def, provider):
self.agent_def = agent_def
self.provider = provider
self.messages = []
async def chat(self, user_input: str) -> str:
# Add user message
# Call LLM
# Return response
```
**Files to Create:**
```
00-chat-loop/
├── README.md
└── src/
├── __init__.py
├── main.py (CLI)
├── agent.py (AgentSession)
├── config.py (basic config)
├── agent_def.py (AGENT.md loader)
├── provider.py (LLM provider)
└── agents/
└── default/
└── AGENT.md
```
**Implementation Notes:**
- Keep AgentSession minimal - no tools, no history persistence yet
- Use litellm for multi-provider support (matches pickle-bot)
- AGENT.md only has `# System Prompt` section
- Config is just YAML with `llm.api_key` and `llm.model`
**Alternative Approaches:**
1. Use langchain instead of direct LLM calls (more abstraction)
2. Use dict config instead of AGENT.md (less file-based)
3. Use click instead of typer (more traditional CLI)
4. Use OpenAI SDK directly instead of litellm (simpler, less flexible)
---
### Step 01: Tools - Agent Can Take Actions
**Problem:** Agent can only talk, can't do anything
**What to Build:**
- ToolRegistry class
- Tool base class
- Basic tools (read, write, bash)
- Tool calling in AgentSession.chat()
**Pickle-bot References:**
- `tools/registry.py` - ToolRegistry
- `tools/base.py` - Tool base class
- `tools/bash.py`, `tools/read.py`, `tools/write.py` - Example tools
- `core/agent.py` - Tool calling logic (tool execution loop)
**Key Code to Copy:**
```python
# From: tools/registry.py
class ToolRegistry:
def __init__(self):
self.tools = {}
def register(self, tool: Tool):
self.tools[tool.name] = tool
async def call(self, name: str, **kwargs):
return await self.tools[name].execute(**kwargs)
```
**Files to Add:**
```
01-tools/
├── src/
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── registry.py
│ │ ├── base.py
│ │ ├── read.py
│ │ ├── write.py
│ │ └── bash.py
│ └── agent.py (modified to call tools)
```
**Implementation Notes:**
- Tools are async functions with JSON schema
- Start with 3 simple tools (read, write, bash)
- Tool calling uses LLM function calling
- Add tool results to message history
- Handle tool call loop (agent may call multiple tools)
**Alternative Approaches:**
1. Use pydantic for tool schemas (more validation)
2. Use decorators for tool registration (less boilerplate)
3. Sync tools instead of async (simpler, less flexible)
4. Use langchain tools (more ecosystem, less control)
---
### Step 02: Skills - Dynamic Capability Loading
**Problem:** Tools are always loaded, need on-demand capabilities
**What to Build:**
- SkillLoader class
- SKILL.md format
- Skill loading tool
- On-demand skill loading
**Pickle-bot References:**
- `core/skills.py` - SkillLoader
- `skills/` directory - Example skills
- `tools/skill_loader.py` - Skill loading tool
**Key Code to Copy:**
```python
# From: core/skills.py
class SkillLoader:
def load(self, skill_name: str) -> Skill:
# Load SKILL.md
# Parse instructions + tools
# Return Skill object
```
**Files to Add:**
```
02-skills/
├── src/
│ ├── skills/
│ │ ├── __init__.py
│ │ ├── loader.py
│ │ └── base.py
│ ├── tools/
│ │ └── skill_loader.py (NEW)
│ └── skills/
│ └── example/
│ └── SKILL.md
```
**Implementation Notes:**
- Skills are loaded on-demand, not at startup
- SKILL.md has instructions + tool definitions
- Skill tool lets agent load skills dynamically
- Keep it simple - no skill dependencies yet
- Skill instructions are injected into system prompt
**Alternative Approaches:**
1. Pre-load all skills at startup (simpler, less dynamic)
2. Use Python modules instead of SKILL.md (more code, less declarative)
3. No skill system, only tools (simpler, less flexible)
---
### Step 03: Persistence - Remember Conversations
**Problem:** Agent forgets everything after restart
**What to Build:**
- HistoryStore class
- JSON-based session storage
- Session recovery on startup
- History manager
**Pickle-bot References:**
- `core/history.py` - HistoryStore
- `core/agent.py` - Session persistence logic
**Key Code to Copy:**
```python
# From: core/history.py
class HistoryStore:
def save_session(self, session_id: str, messages: list):
# Save to JSON file
def load_session(self, session_id: str) -> list:
# Load from JSON file
```
**Files to Add:**
```
03-persistence/
├── src/
│ ├── history.py
│ └── .sessions/
│ └── default.json
```
**Implementation Notes:**
- Use JSON for simplicity (pickle-bot uses same)
- Store in `.sessions/` directory
- Auto-recover last session on startup
- Add session_id to AgentSession
- Keep message history as list of dicts
**Alternative Approaches:**
1. Use SQLite instead of JSON (more structured, more complex)
2. Use pickle instead of JSON (Python-specific, harder to debug)
3. No persistence (simpler, but loses conversations)
---
### Step 04: Slash Commands - User Control
**Problem:** User has no way to control the conversation
**What to Build:**
- CommandRegistry
- Command handler
- Basic commands (/help, /clear, /compact, /reload)
**Pickle-bot References:**
- `core/commands/` - Command system
- `cli/main.py` - Command handling
**Key Code to Copy:**
```python
# From: core/commands/
class CommandRegistry:
def register(self, name: str, handler: Callable):
self.commands[name] = handler
async def handle(self, command: str, session: AgentSession):
await self.commands[command](session)
```
**Implementation Notes:**
- Commands start with /
- Handle in CLI loop before agent chat
- Start with 4-5 basic commands:
- `/help` - Show available commands
- `/clear` - Clear conversation
- `/compact` - Manual compaction
- `/reload` - Reload agent config
- `/exit` - Exit chat
- Easy to add new commands later
**Alternative Approaches:**
1. No command system (simpler, less control)
2. Use CLI flags instead of slash commands (less interactive)
3. Use natural language commands (more flexible, less reliable)
---
### Step 05: Compaction - Handle Long Conversations
**Problem:** Context window fills up, can't have long conversations
**What to Build:**
- ContextGuard class (token counting)
- Message compaction strategy
- Keep recent + important messages
- Compaction triggers
**Pickle-bot References:**
- `core/context_guard.py` - ContextGuard
- `core/agent.py` - Compaction logic
**Key Code to Copy:**
```python
# From: core/context_guard.py
class ContextGuard:
def should_compact(self, messages: list) -> bool:
# Check token count
def compact(self, messages: list) -> list:
# Keep recent + important
```
**Implementation Notes:**
- Use tiktoken for token counting (matches pickle-bot)
- Keep last N messages (simple strategy)
- Summarize old messages (optional, requires LLM call)
- Add /compact command for manual trigger
- Check token count before each LLM call
**Alternative Approaches:**
1. No compaction, just fail (simpler, limited conversations)
2. Summarize all old messages (more context, more tokens)
3. Use sliding window only (no summarization)
4. Use semantic search to find important messages (more complex)
---
### Step 06: Web Tools - Access the Internet
**Problem:** Agent can't search the web or read URLs
**What to Build:**
- WebSearchProvider (Brave Search)
- WebReadProvider (Crawl4AI)
- Web tools (search, read)
**Pickle-bot References:**
- `provider/web_search/` - Search implementation
- `provider/web_read/` - Read implementation
- `tools/web.py` - Web tools
**Implementation Notes:**
- Use Brave Search API (free tier available)
- Use Crawl4AI for reading web pages
- Add web tools to ToolRegistry
- Requires API keys in config
- Handle rate limiting and errors
**Alternative Approaches:**
1. Use Google Search API (more results, requires API key)
2. Use requests + BeautifulSoup for reading (simpler, less robust)
3. No web tools (simpler, less capable)
---
## Phase 2: Event-Driven Architecture
### Step 07: Event-Driven - The Great Refactor
**Problem:** Direct calls make it hard to scale and add features
**What to Build:**
- Event types (InboundEvent, OutboundEvent, EventSource)
- EventBus (pub/sub)
- Base Worker class
- AgentWorker
**Pickle-bot References:**
- `core/events.py` - Event types
- `core/eventbus.py` - EventBus
- `server/worker.py` - Base Worker
- `server/agent_worker.py` - AgentWorker
**Implementation Notes:**
- This is the BIG REFACTOR
- Replace direct chat() calls with events
- Show before/after comparison
- Keep it working throughout refactor
- EventBus uses asyncio.Queue for async processing
- Workers run as background tasks
**Alternative Approaches:**
1. Keep direct calls (simpler, less scalable)
2. Use message queue (Redis, RabbitMQ) (more complex, more scalable)
3. Use callback pattern (simpler, less flexible)
---
### Step 08: Config Hot Reload
**Problem:** Need to restart to test config changes
**What to Build:**
- File watcher (watchdog)
- Reload config.*.yaml on change
- Hot reload trigger
- EventBus integration for reload notifications
**Pickle-bot References:**
- `utils/config_watcher.py` - File watching
- `core/config.py` - Config reloading
**Implementation Notes:**
- Use watchdog library for file watching
- Watch config.*.yaml files for changes
- Reload config on file change
- Emit event on config change via EventBus
- Notify agent of reload in chat
- Keep session history on reload
**Alternative Approaches:**
1. No hot reload, always restart (simpler, slower development)
2. Polling instead of file watcher (simpler, less efficient)
3. Manual reload command only (less convenient)
---
### Step 09: Channels - Multi-Platform Support
**Problem:** Agent only accessible via CLI
**What to Build:**
- Channel abstraction (EventSource subclasses)
- CLI channel
- Telegram channel
- Disk event persistence
**Pickle-bot References:**
- `channel/` - Channel implementations
- `channel/cli.py` - CLI channel
- `channel/telegram.py` - Telegram channel
- `core/eventbus.py` - Event persistence
**Implementation Notes:**
- Each platform is a channel
- Channels emit InboundEvents
- Add disk persistence for events (SQLite or JSON)
- Test with 2 platforms (CLI + Telegram)
- Handle platform-specific formatting
**Alternative Approaches:**
1. CLI only (simpler, less useful)
2. No channel abstraction, separate implementations (less clean)
3. Use bot framework (more abstraction, less control)
---
### Step 10: WebSocket UI - Real-Time Interface
**Problem:** Need visual interface and real-time updates
**What to Build:**
- WebSocketWorker
- WebSocket server
- Simple HTML/JS UI
- Event streaming
**Pickle-bot References:**
- `server/websocket_worker.py` - WebSocket worker
- `api/app.py` - WebSocket endpoints
**Implementation Notes:**
- Use fastapi WebSocket for server
- Stream events to connected clients
- Simple HTML/JS client (single file)
- Show agent status in real-time
- Handle multiple concurrent connections
**Alternative Approaches:**
1. No UI, CLI only (simpler, less visual)
2. Use SSE instead of WebSocket (simpler, less real-time)
3. Build full React/Vue frontend (more complex, more features)
---
## Phase 3: Autonomous & Multi-Agent
### Step 11: Multi-Agent Routing - Right Agent for Right Job
**Problem:** All platforms go to same agent, need specialized agents
**What to Build:**
- Multiple agent definitions
- Routing system
- Binding patterns (regex)
- Source → agent mapping
- Default agent fallback
**Pickle-bot References:**
- `core/routing.py` - Routing logic
- `core/config.py` - Routing config
- Multiple agent definitions in `agents/`
**Implementation Notes:**
- Create multiple agents with different capabilities
- Regex patterns for routing
- Route by source type or content
- Config-based routing rules
- Show multi-agent usage (different agents for different channels)
- Handle routing conflicts
**Alternative Approaches:**
1. Single agent for all channels (simpler, less specialized)
2. Manual agent selection (less automated)
3. Round-robin routing (less intelligent)
---
### Step 12: Cron + Heartbeat - Scheduled Tasks
**Problem:** Agent only responds, never initiates
**What to Build:**
- CronWorker
- CRON.md definitions
- Scheduled agent invocations
- Heartbeat monitoring
**Pickle-bot References:**
- `server/cron_worker.py` - Cron worker
- `core/cron.py` - CRON.md parsing
**Implementation Notes:**
- Use croniter for parsing cron syntax
- CRON.md defines schedules and prompts
- Emit InboundEvent on schedule
- Add heartbeat for health checks
- Store last run times
**Alternative Approaches:**
1. No scheduled tasks (simpler, less autonomous)
2. Use system cron (less portable, less integrated)
3. Use APScheduler (more features, more complexity)
---
### 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
**What to Build:**
- PostMessageBackTool
- Tool that creates OutboundEvent
- Agent can send messages proactively
**Pickle-bot References:**
- `tools/post_message_back.py` - Implementation
**Implementation Notes:**
- Tool creates OutboundEvent
- DeliveryWorker handles it
- Agent can notify user proactively
- Useful for cron task results
- Support multiple channels
**Alternative Approaches:**
1. No proactive messaging (simpler, less autonomous)
2. Use separate notification system (less integrated)
3. Only allow responses to inbound messages (less flexible)
---
### Step 15: Agent Dispatch - Multi-Agent Collaboration
**Problem:** Need specialized agents working together
**What to Build:**
- DispatchEvent
- Subagent tool
- Dispatch result handling
- Agent-to-agent communication
**Pickle-bot References:**
- `core/events.py` - DispatchEvent
- `tools/subagent.py` - Subagent tool
- `server/agent_worker.py` - Dispatch handling
**Implementation Notes:**
- Agent can call other agents via tool
- DispatchEvent → InboundEvent for subagent
- Collect results from subagent
- Show agent collaboration example
- Handle dispatch timeouts
**Alternative Approaches:**
1. Single agent only (simpler, less specialized)
2. Manual agent switching (less automated)
3. No agent-to-agent communication (less collaboration)
---
## Phase 4: Production & Scale
### Step 16: Concurrency Control
**Problem:** Too many concurrent requests overwhelm system
**What to Build:**
- Rate limiting (per agent, per channel)
- Concurrent execution control
- Queue management
- Semaphore-based limiting
**Pickle-bot References:**
- `server/agent_worker.py` - Concurrency control
**Implementation Notes:**
- Use asyncio.Semaphore for limiting
- Per-agent limits (configurable)
- Per-channel limits (configurable)
- Queue management for pending requests
- Show metrics (queue depth, active tasks)
**Alternative Approaches:**
1. No limits (simpler, can overwhelm)
2. Global limit only (less granular)
3. Use external rate limiter (Redis) (more complex, distributed)
---
### Step 17: Memory - Long-Term Knowledge
**Problem:** Agent doesn't remember user preferences or long-term info
**What to Build:**
- Memory structure (topics, projects, daily-notes)
- Memory agent (specialized)
- Memory tools (store/retrieve/search)
- Memory integration
**Pickle-bot References:**
- `tools/memory.py` - Memory tools
- Memory agent definition
**Implementation Notes:**
- Memory agent is specialized for managing knowledge
- Store in structured format (topics, projects, etc.)
- Search and retrieve tools
- Main agent can query memory via dispatch
- Persist memory to disk
**Alternative Approaches:**
1. No memory system (simpler, less personalized)
2. Use database for memory (more structured, more complex)
3. Use vector database for semantic search (more advanced, more complex)
---
## File Organization
```
build-your-own-openclaw/
├── README.md (index)
├── PLAN.md (this file)
├── 00-chat-loop/
│ ├── README.md
│ └── src/
├── 01-tools/
│ ├── README.md
│ └── src/
├── 02-skills/
├── 03-persistence/
├── 04-slash-commands/
├── 05-compaction/
├── 06-web-tools/
├── 07-event-driven/
├── 08-config-hot-reload/
├── 09-channels/
├── 10-websocket/
├── 11-multi-agent-routing/
├── 12-cron-heartbeat/
├── 13-multi-layer-prompts/
├── 14-post-message-back/
├── 15-agent-dispatch/
├── 16-concurrency-control/
└── 17-memory/
├── README.md
└── src/
```
Each step folder contains a snapshot of the codebase at that step.
## Session Workflow
Each step should be implemented in a separate session:
1. **Read this PLAN.md** - Understand what to build
2. **Check pickle-bot** - Find reference implementation
3. **Copy from previous step** - Start with last step's code
4. **Implement the feature** - Following priority order
5. **Write README.md** - Document the step
6. **Test it** - Make sure it runs
7. **Commit** - Save the snapshot
## Success Criteria
Each step is complete when:
- ✅ Code runs without errors
- ✅ Feature works as described
- ✅ README.md explains the step
- ✅ Code matches pickle-bot patterns
- ✅ Incremental from previous step
- ✅ No unnecessary new code
+2 -2
View File
@@ -1,10 +1,10 @@
# Build Your Own OpenClaw
A step-by-step tutorial to build your own AI agent framework, from a simple chat loop to a production-ready multi-agent system.
A step-by-step tutorial to build your own AI agent, from a simple chat loop to a production-ready multi-agent system.
## Overview
**18 progressive steps** that teach you how to build an AI agent framework like pickle-bot, featuring:
**18 progressive steps** that teach you how to build an AI agent like pickle-bot, featuring:
- Tool calling and skill learning
- Multi-platform support (CLI, Telegram, Discord, WebSocket)
- Event-driven architecture
-230
View File
@@ -1,230 +0,0 @@
# Implementation Workflow - Build Your Own OpenClaw
## Overview
This document defines the step-by-step process for implementing each tutorial step. Following this workflow ensures that tutorial code stays aligned with picklebot patterns and avoids unnecessary code drift.
### Core Principles
1. **Copy from picklebot first** - Use battle-tested code as the foundation
2. **Trim future code** - Remove pieces needed only in later steps
3. **Keep it working** - Every step must be runnable
4. **Shared workspace** - Config and workspace files in `default_workspace/` are shared across all steps
### Why "Copy from Picklebot" Over "Merge from Previous Step"?
When a file needs changes, you might be tempted to:
- ❌ Take previous step's code → add features from picklebot → complicated merge
This is hard because:
- Previous step may have drifted from picklebot patterns
- Hard to identify exactly what to add
- Risk of missing picklebot improvements
Instead, always:
- ✅ Take picklebot's code → trim to this step's needs → simpler and cleaner
This works because:
- Picklebot is the source of truth for patterns
- Easier to remove code than to add/merge
- Guarantees alignment with production patterns
---
## Step Implementation Workflow
### Phase 1: Planning (Before Writing Code)
1. **Identify Required Files**
- Read PLAN.md step description
- List all files to be created/modified
- Map each file to picklebot source: `../pickle-bot/src/picklebot/...`
2. **Categorize Each File**
- **Type A**: New file (doesn't exist in previous step)
- **Type B**: Existing file, no new features needed
- **Type C**: Existing file, new features needed
### Phase 2: Implementation (File by File)
**For Type A (New file):**
1. Copy from picklebot → trim to essentials
2. Remove production complexity, keep core logic
**For Type B (Existing, no changes):**
1. Copy from previous step (no changes needed)
2. Verify it still works with new dependencies
**For Type C (Existing, needs changes):**
1. Copy from picklebot (not from previous step)
2. Trim to include ONLY features up to this step
3. Done - no merge needed, picklebot is the source of truth
### Phase 3: Validation
1. Code runs: `uv run my-bot chat`
2. Test the new feature
3. Verify patterns match picklebot
4. Write/update README.md (follow concise format - see "README Format" section below)
---
## How to Trim Picklebot Code
### What to KEEP (Needed for THIS Step)
- ✅ Core logic required for current step's feature
- ✅ Essential dependencies and imports
- ✅ Error handling that affects current functionality
- ✅ Logging/retries/metrics if they're actually used
### What to REMOVE (Future Step Code)
- ❌ Code only needed in later tutorial steps
- ❌ Imports for features not yet implemented
- ❌ Configuration options for future features
- ❌ Helper functions for future capabilities
- ❌ "Forward-looking" abstractions
### Trimming Strategy
1. **Check PLAN.md** - What does THIS step need?
2. **Read picklebot file** - Identify which parts serve this step
3. **Remove future pieces** - Anything not needed until Step X
4. **Keep it working** - Don't break current functionality
### Example
**Step 01 (Tools) trimming:**
- ✅ Keep: ToolRegistry, basic tools (read/write/bash)
- ❌ Remove: SkillLoader (needed in Step 02)
- ❌ Remove: DispatchEvent handling (needed in Step 13)
- ❌ Remove: Concurrency control (needed in Step 16)
---
## Common Scenarios
### Scenario 1: "Picklebot file is very different from previous step"
**When:** Major refactoring happened in picklebot (e.g., Step 08 event-driven refactor)
**Solution:**
1. Copy from picklebot (it's the target architecture)
2. Trim to this step's features
3. Document the refactor in README.md
**Note:** Don't try to "evolve" from previous step - just use picklebot and trim.
### Scenario 2: "Not sure which picklebot file to use"
**When:** Multiple files in picklebot seem relevant
**Solution:**
1. Check PLAN.md - it usually specifies the exact file
2. Look for similar naming in picklebot structure
3. When in doubt, ask or pick the simpler one
### Scenario 3: "Picklebot doesn't have this file"
**When:** Tutorial needs a file that doesn't exist in picklebot
**Solution:**
1. Double-check PLAN.md - are we sure this file is needed?
2. Check if similar functionality exists under different name
3. Only then: write new code (document why in README.md)
### Scenario 4: "Previous step code looks wrong"
**When:** Previous implementation drifted from picklebot patterns
**Solution:**
1. Trust picklebot over previous step
2. Replace with trimmed picklebot version
3. Note the correction in commit message
---
## Quick Reference: Implementing a Step
### Before You Start
- [ ] Read PLAN.md step description
- [ ] List all files needed
- [ ] Map files to picklebot sources
- [ ] Note: `default_workspace/` config files are shared across all steps
### For Each File
- [ ] Determine type: A (new) / B (no changes) / C (needs changes)
- [ ] **Type A**: Copy from picklebot → trim future code
- [ ] **Type B**: Copy from previous step
- [ ] **Type C**: Copy from picklebot → trim to this step (no merge)
### Trimming Checklist
- [ ] Keep only what's needed for THIS step
- [ ] Remove code for future steps
- [ ] Remove unused imports
- [ ] Keep core logic intact
### Validation Checklist
- [ ] `uv run my-bot chat` works
- [ ] New feature works as described
- [ ] Matches picklebot patterns
- [ ] README.md updated (follow concise format from Step 00)
- [ ] Commit with clear message
### README Format
Follow the concise format from Step 00's README:
1. **Title + one-line description**
2. **Prerequisites** (only if needed)
3. **What We will Build?**
- Simple architecture diagram
- Key components (bullet list)
4. **Key Changes**
- Code snippets with file links
- Show only the most important parts
5. **Notes** (optional, only if needed)
6. **How to Run**
- Command + example interaction
7. **What's Next** (link to next step)
**Keep it extremely concise.** No lengthy explanations, no deep dives, no alternatives discussion. The code speaks for itself.
### File Priority Order
1.**picklebot code** (always preferred - trim to essentials)
2.**previous step code** (only if no picklebot reference AND no changes needed)
3.**new code** (only if no reference exists anywhere)
**Key insight:** For files that need changes, always start from picklebot and trim. Don't try to merge previous step with picklebot - picklebot is the source of truth.
---
## How This Connects to PLAN.md
**PLAN.md** tells you **WHAT** to build:
- Which features each step needs
- Which picklebot files to reference
- What the end result should look like
**WORKFLOW.md** tells you **HOW** to build it:
- The process for implementing each file
- How to trim picklebot code
- How to validate your work
### Usage Pattern
1. Read PLAN.md → Understand the step's requirements
2. Follow WORKFLOW.md → Implement file by file
3. Return to PLAN.md → Verify you met the requirements
### Updates to PLAN.md
PLAN.md will reference this workflow in its "Implementation Principles" section:
```markdown
## Implementation Principles
### **CRITICAL: Follow the Workflow**
See [WORKFLOW.md](./WORKFLOW.md) for the step-by-step implementation process.
This ensures consistency with picklebot patterns across all steps.
```