07-event-driven

00-chat-loop/README.md

@@ -49,21 +49,21 @@ class ChatLoop:
 
     async def run(self) -> None:
         """Run the interactive chat loop."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type 'quit' or 'exit' to end the session.\n")
+        self.console.print("Type 'quit' or 'exit' to end the session.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -73,10 +73,10 @@ class ChatLoop:
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 ```
 
 [src/core/agent.py](src/core/agent.py)

00-chat-loop/pyproject.toml

@@ -17,7 +17,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["mybot"]
+packages = ["src/mybot"]
 
 [project.scripts]
 my-bot = "mybot.cli.main:app"

00-chat-loop/src/mybot/cli/chat.py

@@ -5,7 +5,6 @@ import asyncio
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich import print as rprint
 from rich.prompt import Prompt
 from rich.text import Text
 
@@ -38,28 +37,28 @@ class ChatLoop:
 
     def display_agent_response(self, content: str) -> None:
         """Display agent response with styled prefix."""
-        prefix = Text(f"{self.agent.agent_def.id}: ", style="green")
+        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."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type 'quit' or 'exit' to end the session.\n")
+        self.console.print("Type 'quit' or 'exit' to end the session.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -69,10 +68,10 @@ class ChatLoop:
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 
 
 def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None:

01-tools/pyproject.toml

@@ -17,7 +17,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["mybot"]
+packages = ["src/mybot"]
 
 [project.scripts]
 my-bot = "mybot.cli.main:app"

01-tools/src/mybot/cli/chat.py

@@ -5,7 +5,6 @@ import asyncio
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich import print as rprint
 from rich.prompt import Prompt
 from rich.text import Text
 
@@ -38,28 +37,28 @@ class ChatLoop:
 
     def display_agent_response(self, content: str) -> None:
         """Display agent response with styled prefix."""
-        prefix = Text(f"{self.agent.agent_def.id}: ", style="green")
+        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."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type 'quit' or 'exit' to end the session.\n")
+        self.console.print("Type 'quit' or 'exit' to end the session.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -69,10 +68,10 @@ class ChatLoop:
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 
 
 def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None:

02-skills/pyproject.toml

@@ -17,7 +17,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["mybot"]
+packages = ["src/mybot"]
 
 [project.scripts]
 my-bot = "mybot.cli.main:app"

02-skills/src/mybot/cli/chat.py

@@ -5,7 +5,6 @@ import asyncio
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich import print as rprint
 from rich.prompt import Prompt
 from rich.text import Text
 
@@ -38,28 +37,28 @@ class ChatLoop:
 
     def display_agent_response(self, content: str) -> None:
         """Display agent response with styled prefix."""
-        prefix = Text(f"{self.agent.agent_def.id}: ", style="green")
+        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."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type 'quit' or 'exit' to end the session.\n")
+        self.console.print("Type 'quit' or 'exit' to end the session.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -69,10 +68,10 @@ class ChatLoop:
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 
 
 def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None:

03-persistence/pyproject.toml

@@ -17,7 +17,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["mybot"]
+packages = ["src/mybot"]
 
 [project.scripts]
 my-bot = "mybot.cli.main:app"

03-persistence/src/mybot/cli/chat.py

@@ -5,7 +5,6 @@ import asyncio
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich import print as rprint
 from rich.prompt import Prompt
 from rich.text import Text
 
@@ -38,28 +37,28 @@ class ChatLoop:
 
     def display_agent_response(self, content: str) -> None:
         """Display agent response with styled prefix."""
-        prefix = Text(f"{self.agent.agent_def.id}: ", style="green")
+        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."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type 'quit' or 'exit' to end the session.\n")
+        self.console.print("Type 'quit' or 'exit' to end the session.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -69,10 +68,10 @@ class ChatLoop:
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 
 
 

04-slash-commands/README.md

@@ -64,7 +64,7 @@ cmd_response = await self.session.command_registry.dispatch(
     user_input, self.session
 )
 if cmd_response is not None:
-    rprint(cmd_response)
+    self.console.print(cmd_response)
     continue
 
 # Normal chat

04-slash-commands/pyproject.toml

@@ -17,7 +17,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["mybot"]
+packages = ["src/mybot"]
 
 [project.scripts]
 my-bot = "mybot.cli.main:app"

04-slash-commands/src/mybot/cli/chat.py

@@ -5,7 +5,6 @@ import asyncio
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich import print as rprint
 from rich.prompt import Prompt
 from rich.text import Text
 
@@ -38,28 +37,28 @@ class ChatLoop:
 
     def display_agent_response(self, content: str) -> None:
         """Display agent response with styled prefix."""
-        prefix = Text(f"{self.agent.agent_def.id}: ", style="green")
+        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."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type '/help' for commands, 'quit' or 'exit' to end.\n")
+        self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -71,17 +70,17 @@ class ChatLoop:
                         user_input, self.session
                     )
                     if cmd_response is not None:
-                        rprint(cmd_response)
+                        self.console.print(cmd_response)
                         continue
 
                     # Normal chat
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 
 
 

04-slash-commands/src/mybot/core/commands/handlers.py

@@ -25,7 +25,6 @@ class SessionCommand(Command):
             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)
 

05-compaction/pyproject.toml

@@ -17,7 +17,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["mybot"]
+packages = ["src/mybot"]
 
 [project.scripts]
 my-bot = "mybot.cli.main:app"

05-compaction/src/mybot/cli/chat.py

@@ -5,7 +5,6 @@ import asyncio
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich import print as rprint
 from rich.prompt import Prompt
 from rich.text import Text
 
@@ -38,28 +37,28 @@ class ChatLoop:
 
     def display_agent_response(self, content: str) -> None:
         """Display agent response with styled prefix."""
-        prefix = Text(f"{self.agent.agent_def.id}: ", style="green")
+        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."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type '/help' for commands, 'quit' or 'exit' to end.\n")
+        self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -71,17 +70,17 @@ class ChatLoop:
                         user_input, self.session
                     )
                     if cmd_response is not None:
-                        rprint(cmd_response)
+                        self.console.print(cmd_response)
                         continue
 
                     # Normal chat
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 
 
 def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None:

05-compaction/src/mybot/core/commands/handlers.py

@@ -24,7 +24,6 @@ class SessionCommand(Command):
             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)
 

06-web-tools/pyproject.toml

@@ -19,7 +19,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["mybot"]
+packages = ["src/mybot"]
 
 [project.scripts]
 my-bot = "mybot.cli.main:app"

06-web-tools/src/mybot/cli/chat.py

@@ -5,7 +5,6 @@ import asyncio
 import typer
 from rich.console import Console
 from rich.panel import Panel
-from rich import print as rprint
 from rich.prompt import Prompt
 from rich.text import Text
 
@@ -38,28 +37,28 @@ class ChatLoop:
 
     def display_agent_response(self, content: str) -> None:
         """Display agent response with styled prefix."""
-        prefix = Text(f"{self.agent.agent_def.id}: ", style="green")
+        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."""
-        rprint(
+        self.console.print(
             Panel(
                 Text("Welcome to my-bot!", style="bold cyan"),
                 title="Chat",
                 border_style="cyan",
             )
         )
-        rprint("Type '/help' for commands, 'quit' or 'exit' to end.\n")
+        self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n")
 
         try:
             while True:
                 user_input = await asyncio.to_thread(self.get_user_input)
 
                 if user_input.lower() in ("quit", "exit", "q"):
-                    rprint("\n[bold yellow]Goodbye![/bold yellow]")
+                    self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
                     break
 
                 if not user_input:
@@ -71,17 +70,17 @@ class ChatLoop:
                         user_input, self.session
                     )
                     if cmd_response is not None:
-                        rprint(cmd_response)
+                        self.console.print(cmd_response)
                         continue
 
                     # Normal chat
                     response = await self.session.chat(user_input)
                     self.display_agent_response(response)
                 except Exception as e:
-                    rprint(f"\n[bold red]Error:[/bold red] {e}\n")
+                    self.console.print(f"\n[bold red]Error:[/bold red] {e}\n")
 
         except (KeyboardInterrupt, EOFError):
-            rprint("\n[bold yellow]Goodbye![/bold yellow]")
+            self.console.print("\n[bold yellow]Goodbye![/bold yellow]")
 
 
 def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None:

06-web-tools/src/mybot/core/commands/handlers.py

@@ -25,7 +25,6 @@ class SessionCommand(Command):
             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)
 

07-event-driven/README.md

@@ -0,0 +1,140 @@
+# Step 07: Event-Driven Architecture
+
+Replace direct `chat()` calls with event-driven architecture using EventBus and Workers for scalability.
+
+## What We Will Build
+
+```
+┌─────────────┐                    ┌──────────────┐
+│   CLI       │──InboundEvent────▶│  EventBus    │
+│             │                    │              │
+│             │◀──OutboundEvent───│              │
+└─────────────┘                    └──────┬───────┘
+                                          │
+                                          ▼
+                                   ┌──────────────┐
+                                   │ AgentWorker  │
+                                   │              │
+                                   │ - Executes   │
+                                   │   sessions   │
+                                   └──────────────┘
+```
+
+**Key Components:**
+- **EventBus** - Central pub/sub for event distribution
+- **Event Types** - InboundEvent, OutboundEvent
+- **Workers** - Background tasks that process events
+- **AgentWorker** - Handles InboundEvent → executes agent session → emits OutboundEvent
+
+## Key Changes
+
+### 1. Event Types ([src/mybot/core/events.py](src/mybot/core/events.py))
+
+```python
+@dataclass
+class InboundEvent(Event):
+    """Event for external work entering the system."""
+    session_id: str
+    content: str
+    retry_count: int = 0
+
+@dataclass
+class OutboundEvent(Event):
+    """Event for agent responses."""
+    session_id: str
+    content: str
+    error: str | None = None
+```
+
+### 2. EventBus ([src/mybot/core/eventbus.py](src/mybot/core/eventbus.py))
+
+```python
+class EventBus(Worker):
+    def __init__(self, context):
+        self._subscribers: dict[type[Event], list[Handler]] = defaultdict(list)
+        self._queue: asyncio.Queue[Event] = asyncio.Queue()
+
+    async def publish(self, event: Event) -> None:
+        await self._queue.put(event)
+
+    async def run(self) -> None:
+        while True:
+            event = await self._queue.get()
+            await self._dispatch(event)
+```
+
+### 3. AgentWorker ([src/mybot/server/agent_worker.py](src/mybot/server/agent_worker.py))
+
+```python
+class AgentWorker(SubscriberWorker):
+    def __init__(self, context):
+        self.context.eventbus.subscribe(InboundEvent, self.dispatch_event)
+
+    async def dispatch_event(self, event: InboundEvent):
+        agent = Agent(agent_def, self.context)
+        session = agent.resume_session(event.session_id)
+        response = await session.chat(event.content)
+
+        result = OutboundEvent(
+            session_id=event.session_id,
+            content=response,
+        )
+        await self.context.eventbus.publish(result)
+```
+
+### 4. CLI Uses Events ([src/mybot/cli/chat.py](src/mybot/cli/chat.py))
+
+```python
+class ChatLoop:
+    def __init__(self, config: Config):
+        self.context = SharedContext(config)
+        self.workers = [self.context.eventbus, AgentWorker(self.context)]
+        self.context.eventbus.subscribe(OutboundEvent, self.handle_outbound_event)
+
+    async def run(self):
+        for worker in self.workers:
+            worker.start()
+
+        event = InboundEvent(
+            session_id=session_id,
+            content=user_input,
+        )
+        await self.context.eventbus.publish(event)
+
+        response = await self.response_queue.get()
+        self.display_agent_response(response.content)
+```
+
+### 5. SharedContext ([src/mybot/core/context.py](src/mybot/core/context.py))
+
+```python
+class SharedContext:
+    """Global shared state for the application."""
+
+    def __init__(self, config: Config):
+        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.command_registry = CommandRegistry.with_builtins()
+        self.eventbus = EventBus(self)
+```
+
+## How to Run
+
+```bash
+cd 07-event-driven
+uv run my-bot chat
+```
+
+Example interaction:
+```
+You: What tools do you have?
+default: I have access to read, write, and bash tools for file operations
+         and command execution, plus web_search and web_read tools for
+         internet access.
+```
+
+## What's Next
+
+Step 08 will add **Config Hot Reload** - automatically reload configuration when files change without restarting.

07-event-driven/pyproject.toml

@@ -0,0 +1,25 @@
+[project]
+name = "my-bot"
+version = "0.1.0"
+description = "Step 06: Web Tools - Access the Internet"
+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",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/mybot"]
+
+[project.scripts]
+my-bot = "mybot.cli.main:app"

07-event-driven/src/mybot/cli/__init__.py

@@ -0,0 +1,5 @@
+"""CLI interface for my-bot."""
+
+from mybot.cli.main import app
+
+__all__ = ["app"]

07-event-driven/src/mybot/cli/chat.py

@@ -0,0 +1,116 @@
+"""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,
+)
+from mybot.server import (
+    AgentWorker,
+    Worker,
+)
+from mybot.utils.config import Config
+
+
+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)
+
+        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)
+
+    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")
+
+        for worker in self.workers:
+            worker.start()
+
+        session_id = (
+            Agent(self.agent_def, self.context).new_session().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,
+                    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()
+
+
+def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None:
+    """Start interactive chat session."""
+    config = ctx.obj.get("config")
+
+    chat_loop = ChatLoop(config, agent_id=agent_id)
+    asyncio.run(chat_loop.run())

07-event-driven/src/mybot/cli/main.py

@@ -0,0 +1,73 @@
+"""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.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)
+
+
+if __name__ == "__main__":
+    app()

07-event-driven/src/mybot/core/__init__.py

@@ -0,0 +1,6 @@
+"""Core agent components."""
+
+from .agent import Agent, AgentSession
+from .agent_loader import AgentDef, AgentLoader
+
+__all__ = ["Agent", "AgentSession", "AgentDef", "AgentLoader"]

07-event-driven/src/mybot/core/agent.py

@@ -0,0 +1,224 @@
+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.provider.llm import LLMProvider
+from mybot.tools.registry import ToolRegistry
+from mybot.tools.skill_tool import create_skill_tool
+from mybot.tools.websearch_tool import create_websearch_tool
+from mybot.tools.webread_tool import create_webread_tool
+
+from litellm.types.completion import (
+    ChatCompletionMessageParam as Message,
+    ChatCompletionMessageToolCallParam,
+)
+
+if TYPE_CHECKING:
+    from mybot.core.context import SharedContext
+    from mybot.core.agent_loader import AgentDef
+    from mybot.provider.llm import LLMToolCall
+
+
+class Agent:
+    """A configured agent that creates and manages conversation sessions."""
+
+    def __init__(self, agent_def: "AgentDef", context: "SharedContext") -> None:
+        self.agent_def = agent_def
+        self.context = context
+        self.llm = LLMProvider.from_config(agent_def.llm)
+
+    def _build_tools(self) -> ToolRegistry:
+        """Build a ToolRegistry with tools appropriate for the session."""
+        registry = ToolRegistry.with_builtins()
+
+        # Register skill tool if allowed
+        if self.agent_def.allow_skills:
+            skill_tool = create_skill_tool(self.context.skill_loader)
+            if skill_tool:
+                registry.register(skill_tool)
+
+        websearch_tool = create_websearch_tool(self.context)
+        if websearch_tool:
+            registry.register(websearch_tool)
+
+        webread_tool = create_webread_tool(self.context)
+        if webread_tool:
+            registry.register(webread_tool)
+
+        return registry
+
+    def _get_token_threshold(self) -> int:
+        """Get token threshold based on model's context window."""
+        # Default to 80% of 200k context
+        return 160000
+
+    def new_session(
+        self,
+        session_id: str | None = None,
+    ) -> "AgentSession":
+        """Create a new conversation session."""
+        session_id = session_id or str(uuid.uuid4())
+        tools = self._build_tools()
+
+        # Create context guard for this session
+        context_guard = ContextGuard(
+            shared_context=self.context,
+            token_threshold=self._get_token_threshold(),
+        )
+
+        state = SessionState(
+            session_id=session_id,
+            agent=self,
+            messages=[],
+            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)
+        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]
+
+        # Get all messages (no max_history limit)
+        history_messages = self.context.history_store.get_messages(session_id)
+
+        # Convert HistoryMessage to litellm Message format
+        messages: list[Message] = [msg.to_message() for msg in history_messages]
+
+        # Build tools for resumed session
+        tools = self._build_tools()
+
+        # Create context guard
+        context_guard = ContextGuard(
+            shared_context=self.context,
+            token_threshold=self._get_token_threshold(),
+        )
+
+        # Create SessionState with loaded messages
+        state = SessionState(
+            session_id=session_info.id,
+            agent=self,
+            messages=messages,
+            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 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

07-event-driven/src/mybot/core/agent_loader.py

@@ -0,0 +1,77 @@
+"""Agent definition loader."""
+
+from typing import Any
+
+from pydantic import BaseModel, ValidationError
+
+from mybot.utils.config import Config, LLMConfig
+from mybot.utils.def_loader import (
+    DefNotFoundError,
+    InvalidDefError,
+    parse_definition,
+)
+
+
+class AgentDef(BaseModel):
+    """Loaded agent definition with merged settings."""
+
+    id: str
+    name: str
+    description: str = ""
+    agent_md: str
+    llm: LLMConfig
+    allow_skills: bool = False
+
+
+class AgentLoader:
+    """Loads agent definitions from AGENT.md files."""
+
+    @staticmethod
+    def from_config(config: Config) -> "AgentLoader":
+        return AgentLoader(config)
+
+    def __init__(self, config: Config):
+        """Initialize AgentLoader."""
+        self.config = config
+
+    def load(self, agent_id: str) -> AgentDef:
+        """Load agent by ID."""
+        agent_file = self.config.agents_path / agent_id / "AGENT.md"
+        if not agent_file.exists():
+            raise DefNotFoundError("agent", agent_id)
+
+        try:
+            content = agent_file.read_text()
+            agent_def = parse_definition(content, agent_id, self._parse_agent_def)
+        except InvalidDefError:
+            raise
+        except Exception as e:
+            raise InvalidDefError("agent", agent_id, str(e))
+
+        return agent_def
+
+    def _parse_agent_def(
+        self, def_id: str, frontmatter: dict[str, Any], body: str
+    ) -> AgentDef:
+        """Parse agent definition from frontmatter (callback for parse_definition)."""
+        llm_overrides = frontmatter.get("llm")
+        merged_llm = self._merge_llm_config(llm_overrides)
+
+        try:
+            return AgentDef(
+                id=def_id,
+                name=frontmatter["name"],  # type: ignore[misc]
+                description=frontmatter.get("description", ""),
+                agent_md=body.strip(),
+                llm=merged_llm,
+                allow_skills=frontmatter.get("allow_skills", False),
+            )
+        except ValidationError as e:
+            raise InvalidDefError("agent", def_id, str(e))
+
+    def _merge_llm_config(self, agent_llm: dict[str, Any] | None) -> LLMConfig:
+        """Deep merge agent's llm config with global defaults."""
+        base = self.config.llm.model_dump()
+        if agent_llm:
+            base = {**base, **agent_llm}
+        return LLMConfig(**base)

07-event-driven/src/mybot/core/commands/__init__.py

@@ -0,0 +1,6 @@
+"""Slash commands module."""
+
+from mybot.core.commands.base import Command
+from mybot.core.commands.registry import CommandRegistry
+
+__all__ = ["Command", "CommandRegistry"]

07-event-driven/src/mybot/core/commands/base.py

@@ -0,0 +1,20 @@
+"""Base classes for slash commands."""
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from mybot.core.agent import AgentSession
+
+
+class Command(ABC):
+    """Base class for slash commands."""
+
+    name: str
+    aliases: list[str] = []
+    description: str = ""
+
+    @abstractmethod
+    async def execute(self, args: str, session: "AgentSession") -> str:
+        """Execute the command and return response string."""
+        pass

07-event-driven/src/mybot/core/commands/handlers.py

@@ -0,0 +1,108 @@
+"""Built-in slash command handlers."""
+
+from typing import TYPE_CHECKING
+
+from mybot.core.commands.base import Command
+
+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.agent.history_store.history_store.get_session_info(session.session_id)
+
+        # Handle case where session not found in index
+        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)}",
+        ]
+        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.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 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.agent.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.agent.skill_loader.load_skill(skill_id)
+        except FileNotFoundError:
+            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)

07-event-driven/src/mybot/core/commands/registry.py

@@ -0,0 +1,76 @@
+"""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,
+            SkillsCommand,
+            SessionCommand,
+            CompactCommand,
+            ContextCommand,
+        )
+
+        registry = cls()
+        registry.register(HelpCommand())
+        registry.register(SkillsCommand())
+        registry.register(SessionCommand())
+        registry.register(CompactCommand())
+        registry.register(ContextCommand())
+        return registry

07-event-driven/src/mybot/core/context.py

@@ -0,0 +1,25 @@
+from mybot.core.agent_loader import AgentLoader
+from mybot.core.commands.registry import CommandRegistry
+from mybot.core.history import HistoryStore
+from mybot.core.skill_loader import SkillLoader
+from mybot.core.eventbus import EventBus
+from mybot.utils.config import Config
+
+
+class SharedContext:
+    """Global shared state for the application."""
+
+    config: Config
+    history_store: HistoryStore
+    agent_loader: AgentLoader
+    skill_loader: SkillLoader
+    command_registry: CommandRegistry
+    eventbus: EventBus
+
+    def __init__(self, config: Config) -> 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.command_registry = CommandRegistry.with_builtins()
+        self.eventbus = EventBus(self)

07-event-driven/src/mybot/core/context_guard.py

@@ -0,0 +1,145 @@
+"""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
+
+
+# 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 if needed."""
+        token_count = self.estimate_tokens(state)
+
+        if token_count < self.token_threshold:
+            return state
+
+        # First try truncating large tool results
+        state.messages = self._truncate_large_tool_results(state.messages)
+        token_count = self.estimate_tokens(state)
+
+        if token_count < self.token_threshold:
+            return state
+
+        # If still over threshold, compact via summarization
+        return await self._compact_messages(state)
+
+    def _compress_message_count(self, state: "SessionState") -> int:
+        """Calculate how many messages to compress."""
+        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_messages(
+        self,
+        state: "SessionState",
+    ) -> "SessionState":
+        """Compact history by summarizing older messages."""
+        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
+        )
+
+        # Build compacted message list
+        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:])
+
+        # Update state in place
+        state.messages = messages
+        return state

07-event-driven/src/mybot/core/eventbus.py

@@ -0,0 +1,79 @@
+"""Central event bus for pub/sub event distribution."""
+
+import asyncio
+import logging
+from collections import defaultdict
+from typing import Awaitable, Callable, TypeVar
+
+from mybot.server.worker import Worker
+
+from .events import Event
+
+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) -> None:
+        super().__init__(context)
+        self.context = context
+        self._subscribers: dict[type[Event], list[Handler]] = defaultdict(list)
+        self._queue: asyncio.Queue[Event] = asyncio.Queue()
+
+    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")
+
+    async def run(self) -> None:
+        """Process events from queue, starting with recovery."""
+        logger.info("EventBus started")
+
+        # Process events from queue
+        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._notify_subscribers(event)
+        logger.debug(f"Dispatched {event.__class__.__name__} event")
+
+    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}")

07-event-driven/src/mybot/core/events.py

@@ -0,0 +1,71 @@
+"""Event types and data classes for the event bus."""
+
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class Event:
+    """Base class for all typed events."""
+
+    session_id: str
+    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)
+            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
+            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
+
+
+# Registry mapping event class names to event classes
+_EVENT_CLASSES: dict[str, type[Event]] = {
+    "InboundEvent": InboundEvent,
+    "OutboundEvent": OutboundEvent,
+}
+
+
+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)

07-event-driven/src/mybot/core/history.py

@@ -0,0 +1,213 @@
+"""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 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
+    title: str | None = None
+    message_count: int = 0
+    created_at: str
+    updated_at: str
+
+
+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) -> dict[str, Any]:
+        """Create a new conversation session."""
+        now = _now_iso()
+        session = HistorySession(
+            id=session_id,
+            agent_id=agent_id,
+            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

07-event-driven/src/mybot/core/session_state.py

@@ -0,0 +1,35 @@
+"""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
+
+
+@dataclass
+class SessionState:
+    """Pure conversation state + persistence."""
+
+    session_id: str
+    agent: "Agent"
+    messages: list[Message]
+    shared_context: "SharedContext"
+
+    def add_message(self, message: Message) -> None:
+        """Add message to in-memory list + persist."""
+        self.messages.append(message)
+        history_msg = HistoryMessage.from_message(message)
+        self.shared_context.history_store.save_message(self.session_id, history_msg)
+
+    def build_messages(self) -> list[Message]:
+        """Build messages list with system prompt."""
+        system_prompt = self.agent.agent_def.agent_md
+        messages: list[Message] = [{"role": "system", "content": system_prompt}]
+        messages.extend(self.messages)
+        return messages

07-event-driven/src/mybot/core/skill_loader.py

@@ -0,0 +1,69 @@
+"""Skill loader for discovering and loading skills."""
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from pydantic import BaseModel, ConfigDict, ValidationError
+
+from mybot.utils.def_loader import DefNotFoundError, discover_definitions
+
+if TYPE_CHECKING:
+    from mybot.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+
+class SkillDef(BaseModel):
+    """Loaded skill definition."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    id: str
+    name: str
+    description: str
+    content: str
+
+
+class SkillLoader:
+    """Load and manage skill definitions from filesystem."""
+
+    @staticmethod
+    def from_config(config: "Config") -> "SkillLoader":
+        """Create SkillLoader from config."""
+        return SkillLoader(config)
+
+    def __init__(self, config: "Config"):
+        self.config = config
+
+    def discover_skills(self) -> list[SkillDef]:
+        """Scan skills directory and return list of valid SkillDef."""
+        return discover_definitions(
+            self.config.skills_path, "SKILL.md", self._parse_skill_def
+        )
+
+    def _parse_skill_def(
+        self, def_id: str, frontmatter: dict[str, Any], body: str
+    ) -> SkillDef | None:
+        """Parse skill definition from frontmatter (callback for discover_definitions)."""
+        try:
+            return SkillDef(
+                id=def_id,
+                name=frontmatter["name"],  # type: ignore[misc]
+                description=frontmatter["description"],  # type: ignore[misc]
+                content=body.strip(),
+            )
+        except ValidationError as e:
+            logger.warning(f"Invalid skill '{def_id}': {e}")
+            return None
+        except KeyError as e:
+            logger.warning(f"Missing required field in skill '{def_id}': {e}")
+            return None
+
+    def load_skill(self, skill_id: str) -> SkillDef:
+        """Load full skill definition by ID."""
+        skills = self.discover_skills()
+        for skill in skills:
+            if skill.id == skill_id:
+                return skill
+
+        raise DefNotFoundError("skill", skill_id)

07-event-driven/src/mybot/provider/llm/__init__.py

@@ -0,0 +1,5 @@
+"""LLM provider abstraction."""
+
+from .base import LLMProvider, LLMToolCall
+
+__all__ = ["LLMProvider", "LLMToolCall"]

07-event-driven/src/mybot/provider/llm/base.py

@@ -0,0 +1,85 @@
+"""Base LLM provider abstraction."""
+
+from dataclasses import dataclass
+from typing import Any, Optional, cast
+
+from litellm import acompletion, Choices
+from litellm.types.completion import ChatCompletionMessageParam as Message
+
+from mybot.utils.config import LLMConfig
+
+
+@dataclass
+class LLMToolCall:
+    """A tool/function call from the LLM."""
+
+    id: str
+    name: str
+    arguments: str  # JSON string
+
+
+class LLMProvider:
+    """LLM provider using litellm for multi-provider support."""
+
+    def __init__(
+        self,
+        model: str,
+        api_key: str,
+        api_base: Optional[str] = None,
+        temperature: float = 0.7,
+        max_tokens: int = 2048,
+        **kwargs: Any,
+    ):
+        """Initialize LLM provider."""
+        self.model = model
+        self.api_key = api_key
+        self.api_base = api_base
+        self.temperature = temperature
+        self.max_tokens = max_tokens
+        self._settings = kwargs
+
+    @classmethod
+    def from_config(cls, config: LLMConfig) -> "LLMProvider":
+        """Create provider from LLMConfig."""
+        return cls(
+            model=config.model,
+            api_key=config.api_key,
+            api_base=config.api_base,
+            temperature=config.temperature,
+            max_tokens=config.max_tokens,
+        )
+
+    async def chat(
+        self,
+        messages: list[Message],
+        tools: Optional[list[dict[str, Any]]] = None,
+        **kwargs: Any,
+    ) -> tuple[str, list[LLMToolCall]]:
+        """Default implementation using litellm. Subclasses can override."""
+        request_kwargs: dict[str, Any] = {
+            "model": self.model,
+            "messages": messages,
+            "api_key": self.api_key,
+        }
+
+        if self.api_base:
+            request_kwargs["api_base"] = self.api_base
+        if tools:
+            request_kwargs["tools"] = tools
+        request_kwargs.update(kwargs)
+
+        response = await acompletion(**request_kwargs)
+
+        message = cast(Choices, response.choices[0]).message
+
+        return (
+            message.content or "",
+            [
+                LLMToolCall(
+                    id=tc["id"],
+                    name=tc["function"]["name"],
+                    arguments=tc["function"]["arguments"],
+                )
+                for tc in (message.tool_calls or [])
+            ],
+        )

07-event-driven/src/mybot/provider/web_read/__init__.py

@@ -0,0 +1,5 @@
+"""Web read provider module."""
+
+from .base import ReadResult, WebReadProvider
+
+__all__ = ["ReadResult", "WebReadProvider"]

07-event-driven/src/mybot/provider/web_read/base.py

@@ -0,0 +1,41 @@
+"""Base class for web read providers."""
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel
+
+if TYPE_CHECKING:
+    from mybot.utils.config import Config
+
+
+class ReadResult(BaseModel):
+    """Normalized result from reading a web page."""
+
+    url: str
+    title: str
+    content: str  # Markdown content
+    error: str | None = None
+
+
+class WebReadProvider(ABC):
+    """Abstract base class for web page reading providers."""
+
+    @abstractmethod
+    async def read(self, url: str) -> ReadResult:
+        """Read a web page and return normalized content."""
+        pass
+
+    @staticmethod
+    def from_config(config: "Config") -> "WebReadProvider":
+        """Factory method to create provider from config."""
+        if config.webread is None:
+            raise ValueError("Webread not configured")
+
+        match config.webread.provider:
+            case "crawl4ai":
+                from .crawl4ai import Crawl4AIProvider
+
+                return Crawl4AIProvider()
+            case _:
+                raise ValueError(f"Unknown webread provider: {config.webread.provider}")

07-event-driven/src/mybot/provider/web_read/crawl4ai.py

@@ -0,0 +1,36 @@
+"""Crawl4AI provider for web page reading."""
+
+from crawl4ai import AsyncWebCrawler
+
+from .base import WebReadProvider, ReadResult
+
+
+class Crawl4AIProvider(WebReadProvider):
+    """Web read provider using Crawl4AI."""
+
+    def __init__(self):
+        """Initialize Crawl4AI provider."""
+        pass
+
+    async def read(self, url: str) -> ReadResult:
+        """Read a web page using Crawl4AI."""
+        try:
+            async with AsyncWebCrawler(verbose=False) as crawler:
+                result = await crawler.arun(url=url)
+
+                if not result.success:
+                    raise Exception(result.error_message or "Failed to crawl page")
+
+                return ReadResult(
+                    url=url,
+                    title=(result.metadata.get("title", "") if result.metadata else ""),
+                    content=result.markdown or "",
+                    error=None,
+                )
+        except Exception as e:
+            return ReadResult(
+                url=url,
+                title="",
+                content="",
+                error=str(e),
+            )

07-event-driven/src/mybot/provider/web_search/__init__.py

@@ -0,0 +1,5 @@
+"""Web search provider module."""
+
+from .base import SearchResult, WebSearchProvider
+
+__all__ = ["SearchResult", "WebSearchProvider"]

07-event-driven/src/mybot/provider/web_search/base.py

@@ -0,0 +1,42 @@
+"""Base class for web search providers."""
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel
+
+if TYPE_CHECKING:
+    from mybot.utils.config import Config
+
+
+class SearchResult(BaseModel):
+    """Normalized search result from any provider."""
+
+    title: str
+    url: str
+    snippet: str
+
+
+class WebSearchProvider(ABC):
+    """Abstract base class for web search providers."""
+
+    @abstractmethod
+    async def search(self, query: str) -> list[SearchResult]:
+        """Search the web and return normalized results."""
+        pass
+
+    @staticmethod
+    def from_config(config: "Config") -> "WebSearchProvider":
+        """Factory method to create provider from config."""
+        if config.websearch is None:
+            raise ValueError("Websearch not configured")
+
+        match config.websearch.provider:
+            case "brave":
+                from .brave import BraveSearchProvider
+
+                return BraveSearchProvider(config)
+            case _:
+                raise ValueError(
+                    f"Unknown websearch provider: {config.websearch.provider}"
+                )

07-event-driven/src/mybot/provider/web_search/brave.py

@@ -0,0 +1,49 @@
+"""Brave Search API provider."""
+
+from typing import TYPE_CHECKING
+import httpx
+
+from .base import WebSearchProvider, SearchResult
+
+if TYPE_CHECKING:
+    from mybot.utils.config import Config
+
+
+class BraveSearchProvider(WebSearchProvider):
+    """Web search provider using Brave Search API."""
+
+    BASE_URL = "https://api.search.brave.com/res/v1/web/search"
+
+    def __init__(self, config: "Config"):
+        """Initialize Brave Search provider."""
+        self.api_key = config.websearch.api_key
+
+    async def search(self, query: str) -> list[SearchResult]:
+        """Search the web using Brave Search API."""
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                self.BASE_URL,
+                headers={
+                    "Accept": "application/json",
+                    "X-Subscription-Token": self.api_key,
+                },
+                params={
+                    "q": query,
+                    "count": 10,
+                },
+                timeout=30.0,
+            )
+            response.raise_for_status()
+            data = response.json()
+
+        results = []
+        for item in data.get("web", {}).get("results", []):
+            results.append(
+                SearchResult(
+                    title=item.get("title", ""),
+                    url=item.get("url", ""),
+                    snippet=item.get("description", ""),
+                )
+            )
+
+        return results

07-event-driven/src/mybot/server/__init__.py

@@ -0,0 +1,6 @@
+"""Server workers for event-driven architecture."""
+
+from .worker import Worker, SubscriberWorker
+from .agent_worker import AgentWorker
+
+__all__ = ["Worker", "SubscriberWorker", "AgentWorker"]

07-event-driven/src/mybot/server/agent_worker.py

@@ -0,0 +1,123 @@
+"""Agent worker for executing agent jobs."""
+
+import asyncio
+import logging
+from dataclasses import replace
+
+from .worker import SubscriberWorker
+from mybot.core.agent import Agent
+from mybot.core.events import (
+    InboundEvent,
+    OutboundEvent,
+)
+from mybot.utils.def_loader import DefNotFoundError
+
+
+# Maximum number of retry attempts for failed sessions
+MAX_RETRIES = 3
+
+logger = logging.getLogger(__name__)
+
+
+class AgentWorker(SubscriberWorker):
+    """Dispatches events to session executors."""
+
+    def __init__(self, context):
+        super().__init__(context)
+
+        # Auto-subscribe to events
+        self.context.eventbus.subscribe(InboundEvent, self.dispatch_event)
+        self.logger.info("AgentWorker subscribed to InboundEvent events")
+
+    async def dispatch_event(self, event: InboundEvent) -> 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}")
+
+            result_event = OutboundEvent(
+                session_id=event.session_id,
+                content="",
+                error=str(e),
+            )
+            await self.context.eventbus.publish(result_event)
+            return
+
+        asyncio.create_task(self.exec_session(event, agent_def))
+
+    async def exec_session(self, event: InboundEvent, agent_def) -> None:
+        session_id = event.session_id
+
+        try:
+            agent = Agent(agent_def, self.context)
+            if session_id:
+                try:
+                    session = agent.resume_session(session_id)
+                except ValueError:
+                    logger.warning(f"Session {session_id} not found, creating new")
+                    session = agent.new_session(session_id=session_id)
+            else:
+                session = agent.new_session()
+                session_id = session.session_id
+
+            # Check for slash command FIRST
+            if event.content.startswith("/"):
+                result = await self.context.command_registry.dispatch(
+                    event.content, session
+                )
+                if result:
+                    # Emit response and skip agent chat
+                    await self._emit_response(event, result, session, agent_def.id)
+                    logger.info(f"Command completed: {session_id}")
+                    return
+
+            response = await session.chat(event.content)
+            logger.info(f"Session completed: {session_id}")
+
+            result_event = OutboundEvent(
+                session_id=event.session_id,
+                content=response,
+            )
+            await self.context.eventbus.publish(result_event)
+
+        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:
+                result_event = OutboundEvent(
+                    session_id=event.session_id,
+                    content="",
+                    error=str(e),
+                )
+                await self.context.eventbus.publish(result_event)
+
+    async def _emit_response(
+        self,
+        event: InboundEvent,
+        content: str,
+        session,
+        agent_id: str,
+    ) -> None:
+        """Emit response event with content."""
+        result_event = OutboundEvent(
+            session_id=event.session_id,
+            content=content,
+        )
+        await self.context.eventbus.publish(result_event)

07-event-driven/src/mybot/server/worker.py

@@ -0,0 +1,60 @@
+"""Base worker lifecycle management."""
+
+import asyncio
+import logging
+from abc import ABC, abstractmethod
+
+
+class Worker(ABC):
+    """Base class for all workers with lifecycle management."""
+
+    def __init__(self, context):
+        self.context = context
+        self.logger = logging.getLogger(f"mybot.server.{self.__class__.__name__}")
+        self._task: asyncio.Task | None = None
+
+    @abstractmethod
+    async def run(self) -> None:
+        """Main worker loop. Runs until cancelled."""
+        pass
+
+    def start(self) -> asyncio.Task:
+        """Start the worker as an asyncio Task."""
+        self._task = asyncio.create_task(self.run())
+        return self._task
+
+    def is_running(self) -> bool:
+        """Check if worker is actively running."""
+        return self._task is not None and not self._task.done()
+
+    def has_crashed(self) -> bool:
+        """Check if worker crashed (done but not cancelled)."""
+        return (
+            self._task is not None and self._task.done() and not self._task.cancelled()
+        )
+
+    def get_exception(self) -> BaseException | None:
+        """Get the exception if worker crashed, None otherwise."""
+        if self.has_crashed() and self._task is not None:
+            return self._task.exception()
+        return None
+
+    async def stop(self) -> None:
+        """Gracefully stop the worker."""
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+
+
+class SubscriberWorker(Worker):
+    """Worker that only subscribes to events, no active loop."""
+
+    async def run(self) -> None:
+        """Wait for cancellation - actual work happens in event handlers."""
+        try:
+            await asyncio.Future()
+        except asyncio.CancelledError:
+            pass

07-event-driven/src/mybot/tools/__init__.py

@@ -0,0 +1,7 @@
+"""Tools module for agent capabilities."""
+
+from mybot.tools.base import BaseTool, tool
+from mybot.tools.builtin_tools import bash, edit_file, read_file, write_file
+from mybot.tools.registry import ToolRegistry
+
+__all__ = ["BaseTool", "tool", "ToolRegistry", "read_file", "write_file", "edit_file", "bash"]

07-event-driven/src/mybot/tools/base.py

@@ -0,0 +1,63 @@
+"""Base tool interface and decorator."""
+
+import asyncio
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, Callable
+
+if TYPE_CHECKING:
+    from mybot.core.agent import AgentSession
+
+
+class BaseTool(ABC):
+    """Abstract base class for all tools."""
+
+    name: str
+    description: str
+    parameters: dict[str, Any]  # JSON Schema for function calling
+
+    @abstractmethod
+    async def execute(self, session: "AgentSession", **kwargs: Any) -> str:
+        """Execute the tool."""
+
+    def get_tool_schema(self) -> dict[str, Any]:
+        """Get the tool/function schema for LiteLLM."""
+        return {
+            "type": "function",
+            "function": {
+                "name": self.name,
+                "description": self.description,
+                "parameters": self.parameters,
+            },
+        }
+
+
+def tool(name: str, description: str, parameters: dict[str, Any]) -> Callable:
+    """Decorator to register a function as a tool."""
+
+    def decorator(func: Callable) -> "FunctionTool":
+        return FunctionTool(name, description, parameters, func)
+
+    return decorator
+
+
+class FunctionTool(BaseTool):
+    """A tool created from a function using the @tool decorator."""
+
+    def __init__(
+        self,
+        name: str,
+        description: str,
+        parameters: dict[str, Any],
+        func: Callable,
+    ):
+        self.name = name
+        self.description = description
+        self.parameters = parameters
+        self._func = func
+
+    async def execute(self, session: "AgentSession", **kwargs: Any) -> str:
+        """Execute the underlying function."""
+        result = self._func(session=session, **kwargs)
+        if asyncio.iscoroutine(result):
+            result = await result
+        return str(result)

07-event-driven/src/mybot/tools/builtin_tools.py

@@ -0,0 +1,133 @@
+"""Built-in tools for agent capabilities."""
+
+import asyncio
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from mybot.tools.base import tool
+
+if TYPE_CHECKING:
+    from mybot.core.agent import AgentSession
+
+
+# Filesystem tools
+
+
+@tool(
+    name="read",
+    description="Read the contents of a text file",
+    parameters={
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "Path to the file to read"},
+        },
+        "required": ["path"],
+    },
+)
+async def read_file(path: str, session: "AgentSession") -> str:
+    """Read and return the contents of a file at the given path."""
+    try:
+        return Path(path).read_text()
+    except FileNotFoundError:
+        return f"Error: File not found: {path}"
+    except PermissionError:
+        return f"Error: Permission denied reading: {path}"
+    except IsADirectoryError:
+        return f"Error: Path is a directory, not a file: {path}"
+    except Exception as e:
+        return f"Error reading file: {e}"
+
+
+@tool(
+    name="write",
+    description="Write content to a file",
+    parameters={
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "Path to the file to write"},
+            "content": {
+                "type": "string",
+                "description": "Content to write to the file",
+            },
+        },
+        "required": ["path", "content"],
+    },
+)
+async def write_file(path: str, content: str, session: "AgentSession") -> str:
+    """Write content to a file at the given path."""
+    try:
+        Path(path).write_text(content)
+        return f"Successfully wrote to: {path}"
+    except PermissionError:
+        return f"Error: Permission denied writing to: {path}"
+    except IsADirectoryError:
+        return f"Error: Path is a directory, not a file: {path}"
+    except Exception as e:
+        return f"Error writing file: {e}"
+
+
+@tool(
+    name="edit",
+    description="Edit a file by replacing a string with new content",
+    parameters={
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "Path to the file to edit"},
+            "old_text": {"type": "string", "description": "The text to replace"},
+            "new_text": {
+                "type": "string",
+                "description": "The new text to replace with",
+            },
+        },
+        "required": ["path", "old_text", "new_text"],
+    },
+)
+async def edit_file(
+    path: str, old_text: str, new_text: str, session: "AgentSession"
+) -> str:
+    """Edit a file by replacing old_text with new_text."""
+    try:
+        content = Path(path).read_text()
+        if old_text not in content:
+            return f"Error: '{old_text}' not found in {path}"
+        new_content = content.replace(old_text, new_text)
+        Path(path).write_text(new_content)
+        return f"Successfully edited {path}"
+    except FileNotFoundError:
+        return f"Error: File not found: {path}"
+    except PermissionError:
+        return f"Error: Permission denied editing: {path}"
+    except Exception as e:
+        return f"Error editing file: {e}"
+
+
+# Shell tool
+
+
+@tool(
+    name="bash",
+    description="Execute a bash shell command",
+    parameters={
+        "type": "object",
+        "properties": {
+            "command": {"type": "string", "description": "The bash command to execute"},
+        },
+        "required": ["command"],
+    },
+)
+async def bash(command: str, session: "AgentSession") -> str:
+    """Execute a bash command and return the output."""
+    try:
+        process = await asyncio.create_subprocess_shell(
+            command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+        output = stdout.decode() if stdout else ""
+        error = stderr.decode() if stderr else ""
+        if output and error:
+            return f"{output}\n{error}"
+        return output or error or "Command completed with no output"
+    except Exception as e:
+        return f"Error executing command: {e}"

07-event-driven/src/mybot/tools/registry.py

@@ -0,0 +1,56 @@
+"""Tool registry for managing available tools."""
+
+from typing import TYPE_CHECKING, Any
+
+from mybot.tools.base import BaseTool
+from mybot.tools.builtin_tools import bash, edit_file, read_file, write_file
+
+if TYPE_CHECKING:
+    from mybot.core.agent import AgentSession
+
+
+class ToolRegistry:
+    """Registry for all available tools."""
+
+    def __init__(self) -> None:
+        """Initialize an empty tool registry."""
+        self._tools: dict[str, BaseTool] = {}
+
+    def register(self, tool: BaseTool) -> None:
+        """Register a tool."""
+        self._tools[tool.name] = tool
+
+    def get(self, name: str) -> BaseTool | None:
+        """Get a tool by name."""
+        return self._tools.get(name)
+
+    def list_all(self) -> list[BaseTool]:
+        """List all registered tools."""
+        return list(self._tools.values())
+
+    def get_tool_schemas(self) -> list[dict[str, Any]]:
+        """Get tool schemas for all registered tools."""
+        return [tool.get_tool_schema() for tool in self._tools.values()]
+
+    async def execute_tool(
+        self, name: str, session: "AgentSession", **kwargs: Any
+    ) -> str:
+        """Execute a tool by name."""
+        tool = self.get(name)
+        if tool is None:
+            raise ValueError(f"Tool not found: {name}")
+
+        return await tool.execute(session=session, **kwargs)
+
+    @classmethod
+    def with_builtins(cls) -> "ToolRegistry":
+        """Create a ToolRegistry with builtin tools already registered."""
+
+        registry = cls()
+
+        registry.register(read_file)
+        registry.register(write_file)
+        registry.register(edit_file)
+        registry.register(bash)
+
+        return registry

07-event-driven/src/mybot/tools/skill_tool.py

@@ -0,0 +1,51 @@
+"""Skill tool factory for creating dynamic skill tool."""
+
+from typing import TYPE_CHECKING
+
+from mybot.tools.base import tool
+
+if TYPE_CHECKING:
+    from mybot.core.agent import AgentSession
+    from mybot.core.skill_loader import SkillLoader
+
+
+def create_skill_tool(skill_loader: "SkillLoader"):
+    """Factory function to create skill tool with dynamic schema."""
+    skill_metadata = skill_loader.discover_skills()
+
+    if not skill_metadata:
+        return None
+
+    # Build XML description of available skills
+    skills_xml = "<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

07-event-driven/src/mybot/tools/webread_tool.py

@@ -0,0 +1,47 @@
+"""Webread tool factory."""
+
+from typing import TYPE_CHECKING
+
+from mybot.tools.base import BaseTool, tool
+from mybot.provider.web_read import WebReadProvider
+
+if TYPE_CHECKING:
+    from mybot.core.agent import AgentSession
+    from mybot.core.context import SharedContext
+
+
+def create_webread_tool(context: "SharedContext") -> BaseTool | None:
+    """Factory to create webread tool with injected context."""
+    if not context.config.webread:
+        return None
+
+    provider = WebReadProvider.from_config(context.config)
+
+    @tool(
+        name="webread",
+        description=(
+            "Read and extract content from a web page. "
+            "Returns the page content as markdown."
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "url": {
+                    "type": "string",
+                    "description": "The URL to read",
+                }
+            },
+            "required": ["url"],
+        },
+    )
+    async def webread(url: str, session: "AgentSession") -> str:
+        """Read a web page and return markdown content."""
+
+        result = await provider.read(url)
+
+        if result.error:
+            return f"Error reading {url}: {result.error}"
+
+        return f"**{result.title}**\n\n{result.content}"
+
+    return webread

07-event-driven/src/mybot/tools/websearch_tool.py

@@ -0,0 +1,50 @@
+"""Websearch tool factory."""
+
+from typing import TYPE_CHECKING
+
+from mybot.tools.base import BaseTool, tool
+from mybot.provider.web_search import WebSearchProvider
+
+if TYPE_CHECKING:
+    from mybot.core.agent import AgentSession
+    from mybot.core.context import SharedContext
+
+
+def create_websearch_tool(context: "SharedContext") -> BaseTool | None:
+    """Factory to create websearch tool with injected context."""
+    if not context.config.websearch:
+        return None
+
+    provider = WebSearchProvider.from_config(context.config)
+
+    @tool(
+        name="websearch",
+        description=(
+            "Search the web for information. "
+            "Returns a list of results with titles, URLs, and snippets."
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "The search query",
+                }
+            },
+            "required": ["query"],
+        },
+    )
+    async def websearch(query: str, session: "AgentSession") -> str:
+        """Search the web and return formatted results."""
+
+        results = await provider.search(query)
+
+        if not results:
+            return "No results found."
+
+        output = []
+        for i, r in enumerate(results, 1):
+            output.append(f"{i}. **{r.title}**\n   {r.url}\n   {r.snippet}")
+        return "\n\n".join(output)
+
+    return websearch

07-event-driven/src/mybot/utils/config.py

@@ -0,0 +1,81 @@
+"""Configuration management."""
+
+from pathlib import Path
+from typing import Any, Literal
+
+import yaml
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+
+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 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 Config(BaseModel):
+    """Main configuration for step 06."""
+
+    workspace: Path
+    llm: LLMConfig
+    default_agent: str
+    agents_path: Path = Field(default=Path("agents"))
+    skills_path: Path = Field(default=Path("skills"))
+    history_path: Path = Field(default=Path(".sessions"))
+    websearch: BraveWebSearchConfig | None = None
+    webread: Crawl4AIWebReadConfig | 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",
+            "history_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_config(workspace_dir)
+        config_data["workspace"] = workspace_dir
+        return cls.model_validate(config_data)
+
+    @classmethod
+    def _load_config(cls, workspace_dir: Path) -> dict[str, Any]:
+        """Load config from YAML file."""
+        config_file = workspace_dir / "config.user.yaml"
+        if not config_file.exists():
+            raise FileNotFoundError(f"Config file not found: {config_file}")
+
+        with open(config_file) as f:
+            return yaml.safe_load(f) or {}

07-event-driven/src/mybot/utils/def_loader.py

@@ -0,0 +1,105 @@
+"""Shared utilities for loading definition files (agents, skills, crons)."""
+
+import logging
+from pathlib import Path
+from typing import Any, Callable, TypeVar
+
+import yaml
+
+T = TypeVar("T")
+logger = logging.getLogger(__name__)
+
+
+class DefNotFoundError(Exception):
+    """Definition folder or file doesn't exist."""
+
+    def __init__(self, kind: str, def_id: str):
+        super().__init__(f"{kind.capitalize()} not found: {def_id}")
+        self.kind = kind
+        self.def_id = def_id
+
+
+class InvalidDefError(Exception):
+    """Definition file is malformed."""
+
+    def __init__(self, kind: str, def_id: str, reason: str):
+        super().__init__(f"Invalid {kind} '{def_id}': {reason}")
+        self.kind = kind
+        self.def_id = def_id
+        self.reason = reason
+
+
+def parse_definition(
+    content: str,
+    def_id: str,
+    parse_fn: Callable[[str, dict[str, Any], str], T],
+) -> T:
+    """Parse YAML frontmatter + markdown body with type conversion."""
+    # Find frontmatter delimiters
+    if not content.startswith("---\n"):
+        body = content
+        return parse_fn(def_id, {}, body)
+
+    end_delimiter = content.find("\n---\n", 4)
+    if end_delimiter == -1:
+        body = content
+        return parse_fn(def_id, {}, body)
+
+    frontmatter_text = content[4:end_delimiter]
+    body = content[end_delimiter + 5 :]
+
+    raw_dict = yaml.safe_load(frontmatter_text) or {}
+    return parse_fn(def_id, raw_dict, body)
+
+
+def discover_definitions(
+    path: Path,
+    filename: str,
+    parse_fn: Callable[[str, dict[str, Any], str], T | None],
+) -> list[T]:
+    """Scan directory for definition files."""
+    if not path.exists():
+        logger.warning(f"Definitions directory not found: {path}")
+        return []
+
+    results = []
+    for def_dir in path.iterdir():
+        if not def_dir.is_dir():
+            continue
+
+        def_file = def_dir / filename
+        if not def_file.exists():
+            logger.warning(f"No {filename} found in {def_dir.name}")
+            continue
+
+        try:
+            content = def_file.read_text()
+            result = parse_definition(content, def_dir.name, parse_fn)
+            if result is not None:
+                results.append(result)
+        except Exception as e:
+            logger.warning(f"Failed to parse {def_dir.name}: {e}")
+            continue
+
+    return results
+
+
+def write_definition(
+    def_id: str,
+    frontmatter: dict[str, Any],
+    body: str,
+    base_path: Path,
+    filename: str,
+) -> Path:
+    """Write a definition file with YAML frontmatter and markdown body."""
+    def_dir = base_path / def_id
+    def_dir.mkdir(parents=True, exist_ok=True)
+
+    # Build file content with YAML frontmatter
+    yaml_content = yaml.dump(frontmatter, default_flow_style=False, sort_keys=False)
+    content = f"---\n{yaml_content}---\n\n{body.strip()}\n"
+
+    def_file = def_dir / filename
+    def_file.write_text(content)
+
+    return def_file