diff --git a/backend/package/yuxi/channel/events/__init__.py b/backend/package/yuxi/channel/events/__init__.py
new file mode 100644
index 00000000..6aa9a4ba
--- /dev/null
+++ b/backend/package/yuxi/channel/events/__init__.py
@@ -0,0 +1,11 @@
+from yuxi.channel.events.bus import ChannelEventBus, EventHandler, PublishResult, Subscription
+from yuxi.channel.events.types import HOOK_EVENT_TO_TOPIC, EventTopic
+
+__all__ = [
+    "ChannelEventBus",
+    "EventHandler",
+    "EventTopic",
+    "HOOK_EVENT_TO_TOPIC",
+    "PublishResult",
+    "Subscription",
+]
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/events/bus.py b/backend/package/yuxi/channel/events/bus.py
new file mode 100644
index 00000000..2d3af1b6
--- /dev/null
+++ b/backend/package/yuxi/channel/events/bus.py
@@ -0,0 +1,221 @@
+import asyncio
+import bisect
+import fnmatch
+import logging
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+EventHandler = Callable[..., Coroutine[Any, Any, None]]
+_DEFAULT_TIMEOUT = 30.0
+
+
+@dataclass
+class Subscription:
+    callback: EventHandler
+    priority: int = 100
+    enabled: bool = True
+    key: str = ""
+    topic: str = ""
+
+    def __hash__(self) -> int:
+        return id(self.callback)
+
+
+@dataclass
+class PublishResult:
+    topic: str
+    subscriber_count: int = 0
+    error_count: int = 0
+    errors: list[str] = field(default_factory=list)
+
+
+class ChannelEventBus:
+    def __init__(self) -> None:
+        self._subscriptions: dict[str, list[Subscription]] = {}
+
+    def subscribe(
+        self,
+        topic: str,
+        callback: EventHandler,
+        *,
+        priority: int = 100,
+        key: str = "",
+    ) -> Subscription:
+        sub = Subscription(callback=callback, priority=priority, key=key, topic=topic)
+        subs = self._subscriptions.setdefault(topic, [])
+        idx = bisect.bisect_left([s.priority for s in subs], priority)
+        subs.insert(idx, sub)
+        logger.debug("Subscribe: topic=%s key=%s priority=%d", topic, key, priority)
+        return sub
+
+    def unsubscribe(self, topic: str, callback: EventHandler) -> None:
+        subs = self._subscriptions.get(topic)
+        if subs is None:
+            return
+        self._subscriptions[topic] = [s for s in subs if s.callback != callback]
+        if not self._subscriptions[topic]:
+            del self._subscriptions[topic]
+        logger.debug("Unsubscribe: topic=%s remaining=%d", topic, len(self._subscriptions.get(topic, [])))
+
+    def enable(self, topic: str, callback: EventHandler) -> bool:
+        for sub in self._subscriptions.get(topic, []):
+            if sub.callback == callback:
+                sub.enabled = True
+                return True
+        return False
+
+    def disable(self, topic: str, callback: EventHandler) -> bool:
+        for sub in self._subscriptions.get(topic, []):
+            if sub.callback == callback:
+                sub.enabled = False
+                return True
+        return False
+
+    def _match_topics(self, topic: str) -> list[str]:
+        if topic in self._subscriptions:
+            return [topic]
+
+        matched = []
+        for stored in self._subscriptions:
+            if fnmatch.fnmatch(topic, stored) or fnmatch.fnmatch(stored, topic):
+                matched.append(stored)
+        return matched
+
+    async def publish(
+        self,
+        topic: str,
+        *args: Any,
+        timeout: float | None = None,
+        **kwargs: Any,
+    ) -> PublishResult:
+        topics = self._match_topics(topic)
+        if not topics:
+            return PublishResult(topic=topic)
+
+        effective_timeout = timeout if timeout is not None else _DEFAULT_TIMEOUT
+
+        all_subs: list[Subscription] = []
+        for t in topics:
+            for sub in self._subscriptions.get(t, []):
+                if sub.enabled:
+                    all_subs.append(sub)
+
+        all_subs.sort(key=lambda s: s.priority)
+
+        async def _safe_invoke(sub: Subscription) -> str | None:
+            try:
+                await asyncio.wait_for(sub.callback(*args, **kwargs), timeout=effective_timeout)
+                return None
+            except asyncio.TimeoutError:
+                logger.error("EventBus subscriber timed out after %.1fs: topic=%s key=%s", effective_timeout, topic, sub.key)
+                return f"timeout: {sub.key}"
+            except Exception:
+                logger.exception("EventBus subscriber failed: topic=%s key=%s", topic, sub.key)
+                return f"error: {sub.key}"
+
+        results = await asyncio.gather(*(_safe_invoke(s) for s in all_subs))
+
+        errors = [e for e in results if e is not None]
+        return PublishResult(
+            topic=topic,
+            subscriber_count=len(all_subs),
+            error_count=len(errors),
+            errors=errors,
+        )
+
+    async def publish_sequential(
+        self,
+        topic: str,
+        initial: Any = None,
+        *args: Any,
+        timeout: float | None = None,
+        **kwargs: Any,
+    ) -> tuple[Any, PublishResult]:
+        topics = self._match_topics(topic)
+        if not topics:
+            return initial, PublishResult(topic=topic)
+
+        effective_timeout = timeout if timeout is not None else _DEFAULT_TIMEOUT
+
+        all_subs: list[Subscription] = []
+        for t in topics:
+            for sub in self._subscriptions.get(t, []):
+                if sub.enabled:
+                    all_subs.append(sub)
+
+        all_subs.sort(key=lambda s: s.priority)
+
+        result = initial
+        errors: list[str] = []
+        for sub in all_subs:
+            try:
+                next_result = await asyncio.wait_for(sub.callback(result, *args, **kwargs), timeout=effective_timeout)
+                if next_result is not None:
+                    result = next_result
+            except asyncio.TimeoutError:
+                logger.error("EventBus sequential subscriber timed out: topic=%s key=%s", topic, sub.key)
+                errors.append(f"timeout: {sub.key}")
+            except Exception:
+                logger.exception("EventBus sequential subscriber failed: topic=%s key=%s", topic, sub.key)
+                errors.append(f"error: {sub.key}")
+
+        return result, PublishResult(
+            topic=topic,
+            subscriber_count=len(all_subs),
+            error_count=len(errors),
+            errors=errors,
+        )
+
+    async def publish_first(
+        self,
+        topic: str,
+        *args: Any,
+        timeout: float | None = None,
+        **kwargs: Any,
+    ) -> tuple[Any, PublishResult]:
+        topics = self._match_topics(topic)
+        if not topics:
+            return None, PublishResult(topic=topic)
+
+        effective_timeout = timeout if timeout is not None else _DEFAULT_TIMEOUT
+
+        all_subs: list[Subscription] = []
+        for t in topics:
+            for sub in self._subscriptions.get(t, []):
+                if sub.enabled:
+                    all_subs.append(sub)
+
+        all_subs.sort(key=lambda s: s.priority)
+
+        for sub in all_subs:
+            try:
+                result = await asyncio.wait_for(sub.callback(*args, **kwargs), timeout=effective_timeout)
+                if result is not None:
+                    return result, PublishResult(topic=topic, subscriber_count=1)
+            except asyncio.TimeoutError:
+                logger.error("EventBus first subscriber timed out: topic=%s key=%s", topic, sub.key)
+            except Exception:
+                logger.exception("EventBus first subscriber failed: topic=%s key=%s", topic, sub.key)
+
+        return None, PublishResult(topic=topic, subscriber_count=len(all_subs), error_count=len(all_subs),
+                                  errors=[f"all_failed: {s.key}" for s in all_subs])
+
+    def subscriber_count(self, topic: str | None = None) -> int:
+        if topic is not None:
+            return len(self._subscriptions.get(topic, []))
+        return sum(len(subs) for subs in self._subscriptions.values())
+
+    def list_subscriptions(self, topic: str | None = None) -> dict[str, list[Subscription]]:
+        if topic is not None:
+            subs = self._subscriptions.get(topic)
+            return {topic: list(subs)} if subs else {}
+        return {k: list(v) for k, v in self._subscriptions.items()}
+
+    def clear(self) -> None:
+        self._subscriptions.clear()
+
+    def topics(self) -> list[str]:
+        return list(self._subscriptions.keys())
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/events/types.py b/backend/package/yuxi/channel/events/types.py
new file mode 100644
index 00000000..f94feece
--- /dev/null
+++ b/backend/package/yuxi/channel/events/types.py
@@ -0,0 +1,45 @@
+from enum import StrEnum
+
+
+class EventTopic(StrEnum):
+    CHANNEL_STARTING = "channel.starting"
+    CHANNEL_STOPPING = "channel.stopping"
+    CHANNEL_RUNNING = "channel.running"
+    CHANNEL_ERROR = "channel.error"
+    CHANNEL_HEALTH = "channel.health"
+
+    GATEWAY_START = "gateway.start"
+    GATEWAY_STOP = "gateway.stop"
+
+    MESSAGE_RECEIVED = "message.received"
+    MESSAGE_SENDING = "message.sending"
+    MESSAGE_SENT = "message.sent"
+
+    BEFORE_AGENT_RUN = "agent.before_run"
+    AFTER_AGENT_RUN = "agent.after_run"
+    AGENT_BOOTSTRAP = "agent.bootstrap"
+
+    CONFIG_CHANGED = "config.changed"
+    CONFIG_RELOADED = "config.reloaded"
+    CONFIG_WRITE = "config.write"
+
+    SESSION_START = "session.start"
+    SESSION_END = "session.end"
+
+
+HOOK_EVENT_TO_TOPIC: dict[str, str] = {
+    "on_channel_starting": EventTopic.CHANNEL_STARTING,
+    "on_channel_stopping": EventTopic.CHANNEL_STOPPING,
+    "on_channel_running": EventTopic.CHANNEL_RUNNING,
+    "on_channel_error": EventTopic.CHANNEL_ERROR,
+    "on_gateway_start": EventTopic.GATEWAY_START,
+    "on_gateway_stop": EventTopic.GATEWAY_STOP,
+    "on_message_received": EventTopic.MESSAGE_RECEIVED,
+    "on_message_sending": EventTopic.MESSAGE_SENDING,
+    "on_message_sent": EventTopic.MESSAGE_SENT,
+    "on_before_agent_run": EventTopic.BEFORE_AGENT_RUN,
+    "on_after_agent_run": EventTopic.AFTER_AGENT_RUN,
+    "on_agent_bootstrap": EventTopic.AGENT_BOOTSTRAP,
+    "on_session_start": EventTopic.SESSION_START,
+    "on_session_end": EventTopic.SESSION_END,
+}
\ No newline at end of file