feat(channel/runtime): 新增链路追踪与启动运行时工具集

新增trace_ctx实现链路ID上下文管理与日志注入，添加boot.py启动管理器、startup.py启动任务执行器、backoff.py退避策略工具以及trace.py启动链路追踪模块，完善channel运行时基础能力

backend/package/yuxi/channel/runtime/__init__.py

@@ -0,0 +1 @@
+

backend/package/yuxi/channel/runtime/backoff.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import random
+import time
+from dataclasses import dataclass
+from typing import Any
+from collections.abc import Awaitable, Callable
+
+logger = logging.getLogger(__name__)
+
+BackoffFn = Callable[[int], float]
+
+
+@dataclass
+class BackoffConfig:
+    base_delay: float = 5.0
+    max_delay: float = 300.0
+    exponent: float = 2.0
+    jitter: bool = True
+    jitter_factor: float = 0.1
+    max_retries: int = 10
+
+    def compute_delay(self, attempt: int) -> float:
+        delay = self.base_delay * (self.exponent**attempt)
+        delay = min(delay, self.max_delay)
+        if self.jitter:
+            jitter_amount = delay * self.jitter_factor
+            delay += random.uniform(-jitter_amount, jitter_amount)
+            delay = max(delay, 0.01)
+        return delay
+
+
+class ErrorBackoff:
+    """错误退避控制器 — 管理重试逻辑与指数退避。
+
+    使用按 key 分锁策略，不同 key 之间的退避操作互不阻塞，
+    避免粗粒度单锁成为高并发场景下的全局瓶颈。
+    """
+
+    def __init__(self, config: BackoffConfig | None = None):
+        self.config = config or BackoffConfig()
+        self._attempts: dict[str, int] = {}
+        self._last_error: dict[str, float] = {}
+        self._locks: dict[str, asyncio.Lock] = {}
+        self._global_lock = asyncio.Lock()
+
+    def _get_lock(self, key: str) -> asyncio.Lock:
+        """获取指定 key 的独立锁，必要时惰性创建。"""
+        lock = self._locks.get(key)
+        if lock is None:
+            lock = asyncio.Lock()
+            self._locks[key] = lock
+        return lock
+
+    async def execute(
+        self,
+        key: str,
+        operation: Callable[[], Awaitable[Any]],
+        *,
+        on_retry: Callable[[int, Exception], Awaitable[None]] | None = None,
+    ) -> Any:
+        from yuxi.channel.errors import _infer_severity
+        from yuxi.channel.protocols import ErrorSeverity
+
+        lock = self._get_lock(key)
+        async with lock:
+            attempt = self._attempts.get(key, 0)
+
+            for i in range(self.config.max_retries + 1):
+                try:
+                    result = await operation()
+                    self._attempts[key] = 0
+                    self._last_error.pop(key, None)
+                    return result
+                except Exception as e:
+                    self._attempts[key] = attempt + i + 1
+                    self._last_error[key] = time.time()
+
+                    if _infer_severity(e) == ErrorSeverity.FATAL:
+                        logger.error("Operation '%s' failed with fatal error: %s", key, e)
+                        raise
+
+                    if i >= self.config.max_retries:
+                        logger.error("Operation '%s' failed after %d retries: %s", key, self.config.max_retries, e)
+                        raise
+
+                    delay = self.config.compute_delay(attempt + i)
+                    logger.warning(
+                        "Operation '%s' failed (attempt %d/%d), retrying in %.2fs: %s",
+                        key,
+                        i + 1,
+                        self.config.max_retries + 1,
+                        delay,
+                        e,
+                    )
+
+                    if on_retry:
+                        await on_retry(i, e)
+
+                    await asyncio.sleep(delay)
+
+    def get_attempts(self, key: str) -> int:
+        return self._attempts.get(key, 0)
+
+    def reset(self, key: str) -> None:
+        self._attempts.pop(key, None)
+        self._last_error.pop(key, None)
+        self._locks.pop(key, None)
+
+    def is_backing_off(self, key: str, cooldown_seconds: float = 300.0) -> bool:
+        last_error = self._last_error.get(key)
+        if last_error is None:
+            return False
+        return time.time() - last_error < cooldown_seconds
+
+    def get_backoff_remaining(self, key: str, cooldown_seconds: float = 300.0) -> float:
+        last_error = self._last_error.get(key)
+        if last_error is None:
+            return 0.0
+        remaining = cooldown_seconds - (time.time() - last_error)
+        return max(remaining, 0.0)

backend/package/yuxi/channel/runtime/boot.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Awaitable, Callable
+
+from .startup import StartupTaskRunner
+
+logger = logging.getLogger(__name__)
+
+StartupTask = Callable[[], Awaitable[Any]]
+
+
+class BootManager:
+    """启动管理器 — 协调系统启动阶段的任务执行顺序。"""
+
+    def __init__(self):
+        self._phases: dict[str, list[tuple[int, StartupTask]]] = {
+            "early": [],
+            "config": [],
+            "plugins": [],
+            "channels": [],
+            "services": [],
+            "late": [],
+        }
+        self._runner = StartupTaskRunner()
+
+    def register(self, phase: str, task: StartupTask, *, priority: int = 0) -> None:
+        """注册一个启动任务到指定阶段。
+
+        priority 值越大优先级越高，同阶段内优先执行。
+        """
+        if phase not in self._phases:
+            raise ValueError(f"Unknown boot phase: {phase}")
+        self._phases[phase].append((priority, task))
+
+    async def run_startup_tasks(self) -> None:
+        """按阶段顺序执行所有启动任务，阶段内按 priority 降序执行。"""
+        for phase, tasks in self._phases.items():
+            if not tasks:
+                continue
+            ordered = sorted(tasks, key=lambda x: x[0], reverse=True)
+            logger.info("Boot phase: %s (%d tasks)", phase, len(ordered))
+            for _, task in ordered:
+                try:
+                    await self._runner.run(task)
+                except Exception:
+                    logger.exception("Boot task failed in phase '%s'", phase)
+                    raise
+
+    def get_phases(self) -> dict[str, list[StartupTask]]:
+        return {k: [t for _, t in v] for k, v in self._phases.items()}
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "phases": {
+                phase: [getattr(t, "__name__", str(t)) for t in tasks]
+                for phase, tasks in self.get_phases().items()
+            },
+            "total_tasks": sum(len(v) for v in self._phases.values()),
+        }

backend/package/yuxi/channel/runtime/startup.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable
+
+from yuxi.channel.runtime.backoff import BackoffConfig
+
+logger = logging.getLogger(__name__)
+
+StartupTask = Callable[[], Awaitable[Any]]
+
+
+@dataclass
+class StartupTaskResult:
+    name: str
+    success: bool
+    duration_ms: float
+    error: str | None = None
+    skipped: bool = False
+
+
+_StartupBackoffConfig = BackoffConfig(
+    base_delay=0.5,
+    max_delay=5.0,
+    exponent=1.5,
+    jitter=True,
+    jitter_factor=0.1,
+    max_retries=3,
+)
+
+
+class StartupTaskRunner:
+    """启动任务执行器 — 支持超时、重试、并行执行与结果追踪。"""
+
+    def __init__(self, default_timeout: float = 30.0, max_retries: int = 2, backoff_config: BackoffConfig | None = None):
+        self.default_timeout = default_timeout
+        self.max_retries = max_retries
+        self._backoff_config = backoff_config or _StartupBackoffConfig
+        self._results: list[StartupTaskResult] = []
+        self._results_lock = asyncio.Lock()
+
+    async def run(
+        self,
+        task: StartupTask,
+        *,
+        name: str | None = None,
+        timeout: float | None = None,
+        retries: int | None = None,
+    ) -> StartupTaskResult:
+        task_name = name or getattr(task, "__name__", "unknown")
+        timeout_val = timeout if timeout is not None else self.default_timeout
+        retry_count = retries if retries is not None else self.max_retries
+
+        start = time.time()
+        last_error: Exception | None = None
+
+        for attempt in range(retry_count + 1):
+            try:
+                await asyncio.wait_for(task(), timeout=timeout_val)
+                duration_ms = (time.time() - start) * 1000
+                result = StartupTaskResult(
+                    name=task_name,
+                    success=True,
+                    duration_ms=duration_ms,
+                )
+                async with self._results_lock:
+                    self._results.append(result)
+                logger.info("Startup task '%s' completed in %.2fms", task_name, duration_ms)
+                return result
+            except asyncio.TimeoutError:
+                last_error = asyncio.TimeoutError(f"Task '{task_name}' timed out after {timeout_val}s")
+                logger.warning("Startup task '%s' timed out (attempt %d/%d)", task_name, attempt + 1, retry_count + 1)
+            except Exception as e:
+                last_error = e
+                logger.exception("Startup task '%s' failed (attempt %d/%d)", task_name, attempt + 1, retry_count + 1)
+                if attempt < retry_count:
+                    await asyncio.sleep(self._backoff_config.compute_delay(attempt))
+
+        duration_ms = (time.time() - start) * 1000
+        result = StartupTaskResult(
+            name=task_name,
+            success=False,
+            duration_ms=duration_ms,
+            error=str(last_error) if last_error else "Unknown error",
+        )
+        async with self._results_lock:
+            self._results.append(result)
+        return result
+
+    async def run_parallel(
+        self,
+        tasks: list[tuple[StartupTask, str]],
+        *,
+        timeout: float | None = None,
+    ) -> list[StartupTaskResult]:
+        """并行执行多个启动任务。"""
+        coros = [self.run(task, name=name, timeout=timeout) for task, name in tasks]
+        return await asyncio.gather(*coros, return_exceptions=True)
+
+    async def get_results(self) -> list[StartupTaskResult]:
+        async with self._results_lock:
+            return list(self._results)
+
+    async def clear_results(self) -> None:
+        async with self._results_lock:
+            self._results.clear()

backend/package/yuxi/channel/runtime/trace.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import time as _time
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any
+
+
+class PhaseStatus(StrEnum):
+    PENDING = "pending"
+    RUNNING = "running"
+    OK = "ok"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+
+
+class GatewayReadyState(StrEnum):
+    NOT_READY = "not_ready"
+    READY = "ready"
+    DEGRADED = "degraded"
+    FAILED = "failed"
+
+
+@dataclass
+class PhaseMark:
+    name: str
+    status: PhaseStatus = PhaseStatus.PENDING
+    started_at: float | None = None
+    finished_at: float | None = None
+
+    @property
+    def elapsed_ms(self) -> float | None:
+        if self.started_at and self.finished_at:
+            return (self.finished_at - self.started_at) * 1000
+        return None
+
+
+@dataclass
+class StartupTrace:
+    """启动追踪器 — 记录系统启动各阶段耗时。"""
+
+    started_at: float = 0.0
+    completed_at: float = 0.0
+    failed_at: float = 0.0
+    error: str | None = None
+    phases: list[dict[str, Any]] = field(default_factory=list)
+
+    _phase_marks: list[PhaseMark] = field(default_factory=list, repr=False)
+    _active_phase: str | None = field(default=None, repr=False)
+    ready_state: GatewayReadyState = GatewayReadyState.NOT_READY
+    trace_id: str = ""
+
+    def start(self) -> None:
+        self.started_at = _time.time()
+        self.phases.clear()
+
+    def add_phase(self, name: str, duration_ms: float, *, success: bool = True, error: str | None = None) -> None:
+        self.phases.append(
+            {
+                "name": name,
+                "duration_ms": round(duration_ms, 2),
+                "success": success,
+                "error": error,
+            }
+        )
+
+    def complete(self) -> None:
+        self.completed_at = _time.time()
+
+    def fail(self, error: str) -> None:
+        self.failed_at = _time.time()
+        self.error = error
+
+    @property
+    def duration(self) -> float:
+        if self.completed_at:
+            return self.completed_at - self.started_at
+        if self.failed_at:
+            return self.failed_at - self.started_at
+        return _time.time() - self.started_at
+
+    @property
+    def success(self) -> bool:
+        return self.completed_at > 0 and self.failed_at == 0
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "trace_id": self.trace_id,
+            "started_at": self.started_at,
+            "completed_at": self.completed_at,
+            "failed_at": self.failed_at,
+            "duration_seconds": round(self.duration, 3),
+            "success": self.success,
+            "ready_state": self.ready_state.value,
+            "error": self.error,
+            "phases": list(self.phases),
+        }
+
+    def start_phase(self, name: str) -> None:
+        mark = PhaseMark(name=name, status=PhaseStatus.RUNNING, started_at=_time.time())
+        self._upsert_mark(mark)
+        self._active_phase = name
+
+    def complete_phase(self, name: str, ok: bool = True, error: str | None = None) -> None:
+        mark = self._find_mark(name)
+        if mark is None:
+            return
+        mark.status = PhaseStatus.OK if ok else PhaseStatus.FAILED
+        mark.finished_at = _time.time()
+        self._active_phase = None
+        duration_ms = mark.elapsed_ms or 0
+        self.add_phase(name, duration_ms, success=ok, error=error)
+
+    def skip_phase(self, name: str) -> None:
+        mark = self._find_mark(name)
+        if mark is not None:
+            mark.status = PhaseStatus.SKIPPED
+        else:
+            mark = PhaseMark(name=name, status=PhaseStatus.SKIPPED)
+            self._phase_marks.append(mark)
+        self._active_phase = None
+
+    def finalize(self, ready_state: GatewayReadyState) -> None:
+        self.ready_state = ready_state
+        if not self.completed_at and not self.failed_at:
+            self.completed_at = _time.time()
+
+    def summary(self) -> dict[str, Any]:
+        return {
+            "trace_id": self.trace_id,
+            "ready_state": self.ready_state.value,
+            "total_elapsed_ms": round(self.duration * 1000, 2) if self.duration else None,
+            "phases": [
+                {
+                    "name": p.name,
+                    "status": p.status.value,
+                    "elapsed_ms": round(p.elapsed_ms, 2) if p.elapsed_ms else None,
+                }
+                for p in self._phase_marks
+            ],
+            "errors": [
+                e["error"] for e in self.phases if not e.get("success") and e.get("error")
+            ],
+        }
+
+    def _upsert_mark(self, mark: PhaseMark) -> None:
+        existing = self._find_mark(mark.name)
+        if existing:
+            idx = self._phase_marks.index(existing)
+            self._phase_marks[idx] = mark
+        else:
+            self._phase_marks.append(mark)
+
+    def _find_mark(self, name: str) -> PhaseMark | None:
+        for p in self._phase_marks:
+            if p.name == name:
+                return p
+        return None

backend/package/yuxi/channel/runtime/trace_ctx.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import contextvars
+import logging
+import uuid
+from collections.abc import Generator
+from contextlib import contextmanager
+from typing import Any
+
+
+_trace_id_var: contextvars.ContextVar[str | None] = contextvars.ContextVar(
+    "channel_trace_id", default=None
+)
+
+
+class TraceIdFilter(logging.Filter):
+    """日志过滤器：自动将当前 trace_id 注入 LogRecord 的 trace_id 属性。
+
+    配合 logging.Formatter 使用，在日志格式中包含 %(trace_id)s 即可。
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        trace_id = get_trace_id()
+        record.trace_id = trace_id or "-"
+        return True
+
+
+def generate_trace_id() -> str:
+    return uuid.uuid4().hex[:16]
+
+
+def get_trace_id() -> str | None:
+    return _trace_id_var.get()
+
+
+def set_trace_id(trace_id: str | None) -> None:
+    _trace_id_var.set(trace_id)
+
+
+@contextmanager
+def trace_context(trace_id: str | None = None, **metadata: Any) -> Generator[str, None, None]:
+    token = _trace_id_var.set(trace_id or generate_trace_id())
+    try:
+        yield _trace_id_var.get()
+    finally:
+        _trace_id_var.reset(token)
+
+
+def install_trace_filter(logger_name: str | None = None) -> None:
+    root_logger = logging.getLogger(logger_name)
+    if not any(isinstance(f, TraceIdFilter) for f in root_logger.filters):
+        root_logger.addFilter(TraceIdFilter())