release: bump version to 0.38.0

2026-04-23 17:57:35 +08:00
parent 195a8bf71c
commit d5f3784ffb
39 changed files with 4958 additions and 146 deletions
--- a/backend/app/api/v1/system_control.py
+++ b/backend/app/api/v1/system_control.py
@@ -4,12 +4,15 @@ import os
 import subprocess
 import sys

-from fastapi import APIRouter, Depends, HTTPException, status
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from pydantic import BaseModel

 from app.core.config import ROOT_DIR
 from app.core.security import get_current_user
 from app.models.user import User
+from app.services.persistent_logs import record_audit_log, record_system_log
 from app.services.system_control import (
    build_task_id,
    clear_active_task_id,
@@ -23,6 +26,15 @@ from app.services.system_control import (
    set_active_task_id,
    upsert_task_state,
 )
+from app.services.system_logs import (
+    DEFAULT_LOG_LINE_LIMIT,
+    MAX_LOG_LINE_LIMIT,
+    SUPPORTED_LOG_LEVELS,
+    append_buffer_log,
+    list_log_sources,
+    normalize_log_level,
+    read_log_snapshot,
+)

 router = APIRouter()

@@ -47,6 +59,59 @@ class RestartTaskLogsResponse(BaseModel):
    lines: list[str]


+class SystemLogSourceSummary(BaseModel):
+    source_id: str
+    name: str
+    kind: str
+    location: str
+    description: str
+    category: str
+    status: str
+
+
+class SystemLogSourcesResponse(BaseModel):
+    items: list[SystemLogSourceSummary]
+
+
+class SystemLogDailyMarker(BaseModel):
+    date_token: str
+    total: int
+    dominant_level: str
+
+
+class SystemLogSnapshotResponse(BaseModel):
+    source_id: str
+    name: str
+    kind: str
+    location: str
+    description: str
+    category: str
+    status: str
+    level: str
+    selected_levels: list[str] = []
+    search_query: str = ""
+    available_levels: list[str]
+    daily_markers: list[SystemLogDailyMarker] = []
+    line_limit: int
+    line_count: int
+    lines: list[str]
+
+
+class EarthClientLogEventCreate(BaseModel):
+    level: str = "error"
+    message: str
+    category: str | None = None
+    url: str | None = None
+    module: str | None = None
+    detail: str | None = None
+
+
+class EarthClientLogEventResponse(BaseModel):
+    accepted: bool
+    source_id: str
+    level: str
+
+
 def ensure_super_admin(current_user: User) -> None:
    if not require_super_admin(current_user.role):
        raise HTTPException(
@@ -55,9 +120,22 @@ def ensure_super_admin(current_user: User) -> None:
        )


+def validate_log_date(raw_value: str | None, field_name: str) -> str | None:
+    if raw_value in {None, ""}:
+        return None
+    try:
+        return datetime.strptime(raw_value, "%Y-%m-%d").date().isoformat()
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"{field_name} must be in YYYY-MM-DD format",
+        ) from exc
+
+
@router.post("/restart-tasks", response_model=RestartTaskResponse)
 async def create_restart_task(
    payload: RestartTaskCreate,
+    request: Request,
    current_user: User = Depends(get_current_user),
 ):
    ensure_super_admin(current_user)
@@ -133,11 +211,31 @@ async def create_restart_task(
            requested_by=requested_by,
        )
        clear_active_task_id(task_id)
+        await record_audit_log(
+            action="system.restart_task.requested",
+            actor_id=current_user.id,
+            actor_name=current_user.username,
+            target_type="restart_task",
+            target_id=task_id,
+            result="failed",
+            ip=request.client.host if request.client else None,
+            details={"action": payload.action, "message": task_state["message"]},
+        )
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=task_state["message"],
        ) from exc

+    await record_audit_log(
+        action="system.restart_task.requested",
+        actor_id=current_user.id,
+        actor_name=current_user.username,
+        target_type="restart_task",
+        target_id=task_id,
+        result="accepted",
+        ip=request.client.host if request.client else None,
+        details={"action": payload.action},
+    )
    return task_state


@@ -165,3 +263,92 @@ async def get_restart_task_logs(
    if task is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Restart task not found")
    return {"task_id": task_id, "lines": get_task_logs(task_id)}
+
+
+@router.get("/logs/sources", response_model=SystemLogSourcesResponse)
+async def get_system_log_sources(
+    current_user: User = Depends(get_current_user),
+):
+    ensure_super_admin(current_user)
+    return {"items": list_log_sources()}
+
+
+@router.get("/logs/{source_id}", response_model=SystemLogSnapshotResponse)
+async def get_system_log_snapshot(
+    source_id: str,
+    limit: int = DEFAULT_LOG_LINE_LIMIT,
+    level: str = "all",
+    levels: str | None = Query(None, description="Comma-separated log levels"),
+    start_date: str | None = Query(None, description="Filter logs from this date (YYYY-MM-DD)"),
+    end_date: str | None = Query(None, description="Filter logs until this date (YYYY-MM-DD)"),
+    search: str | None = Query(None, description="Case-insensitive substring search"),
+    current_user: User = Depends(get_current_user),
+):
+    ensure_super_admin(current_user)
+
+    if limit < 1 or limit > MAX_LOG_LINE_LIMIT:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"limit must be between 1 and {MAX_LOG_LINE_LIMIT}",
+        )
+    if str(level).strip().lower() not in SUPPORTED_LOG_LEVELS and normalize_log_level(level) == "all" and str(level).strip().lower() not in {"", "all"}:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported log level")
+    if levels:
+        for raw_level in str(levels).split(","):
+            normalized_level = str(raw_level).strip().lower()
+            if not normalized_level:
+                continue
+            if normalized_level not in SUPPORTED_LOG_LEVELS and normalize_log_level(normalized_level) == "all":
+                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported log level")
+    normalized_start_date = validate_log_date(start_date, "start_date")
+    normalized_end_date = validate_log_date(end_date, "end_date")
+    if normalized_start_date and normalized_end_date and normalized_start_date > normalized_end_date:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="start_date must be earlier than or equal to end_date")
+
+    snapshot = read_log_snapshot(
+        source_id,
+        limit,
+        level=level,
+        levels=levels,
+        start_date=normalized_start_date,
+        end_date=normalized_end_date,
+        search=search,
+    )
+    if snapshot is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Log source not found")
+    return snapshot
+
+
+@router.post("/logs/earth-client", response_model=EarthClientLogEventResponse)
+async def ingest_earth_client_log(
+    payload: EarthClientLogEventCreate,
+    request: Request,
+):
+    normalized_level = normalize_log_level(payload.level)
+    append_buffer_log(
+        "earth-client",
+        level=normalized_level,
+        message=payload.message,
+        context={
+            "category": payload.category or "",
+            "url": payload.url or "",
+            "module": payload.module or "",
+            "detail": payload.detail or "",
+        },
+    )
+    await record_system_log(
+        source="earth-client",
+        service="earth",
+        module=payload.module or "earth-client",
+        event="earth.client.runtime_log",
+        level=normalized_level,
+        message=payload.message,
+        category=payload.category or "client-runtime",
+        context={
+            "url": payload.url or "",
+            "detail": payload.detail or "",
+            "module": payload.module or "",
+            "client_ip": request.client.host if request.client else "",
+        },
+    )
+    return {"accepted": True, "source_id": "earth-client", "level": normalized_level}
--- a/backend/app/api/v1/visualization.py
+++ b/backend/app/api/v1/visualization.py
@@ -5,6 +5,7 @@ Returns GeoJSON format compatible with Three.js, CesiumJS, and Unreal Cesium.
 """

 from datetime import UTC, datetime
+import logging
 import math
 import httpx
 from fastapi import APIRouter, HTTPException, Depends, Query, Response
@@ -23,8 +24,10 @@ from app.models.collected_data import CollectedData
 from app.services.bgp_collectors import build_bgp_collector_coverage
 from app.services.cable_graph import build_graph_from_data, CableGraph, haversine_distance
 from app.services.collectors.bgp_common import RIPE_RIS_COLLECTOR_COORDS
+from app.services.persistent_logs import record_system_log

 router = APIRouter()
+logger = logging.getLogger(__name__)
 TERRAIN_TILE_URL_TEMPLATE = (
    "https://s3.amazonaws.com/elevation-tiles-prod/terrarium/{z}/{x}/{y}.png"
 )
@@ -990,6 +993,17 @@ async def get_cables_geojson(db: AsyncSession = Depends(get_db)):
    except HTTPException:
        raise
    except Exception as e:
+        logger.exception("Failed to build cables GeoJSON response")
+        await record_system_log(
+            source="backend",
+            service="api",
+            module=__name__,
+            event="visualization.cables.load_failed",
+            level="error",
+            message="Failed to build cables GeoJSON response",
+            category="visualization",
+            context={"error": str(e)},
+        )
        raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")


@@ -1026,6 +1040,17 @@ async def get_landing_points_geojson(db: AsyncSession = Depends(get_db)):
    except HTTPException:
        raise
    except Exception as e:
+        logger.exception("Failed to build landing points GeoJSON response")
+        await record_system_log(
+            source="backend",
+            service="api",
+            module=__name__,
+            event="visualization.landing_points.load_failed",
+            level="error",
+            message="Failed to build landing points GeoJSON response",
+            category="visualization",
+            context={"error": str(e)},
+        )
        raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")


--- a/backend/app/core/request_context.py
+++ b/backend/app/core/request_context.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from contextvars import ContextVar
+
+
+request_id_context: ContextVar[str | None] = ContextVar("request_id", default=None)
+
+
+def set_request_id(request_id: str | None) -> None:
+    request_id_context.set(request_id)
+
+
+def get_request_id() -> str | None:
+    return request_id_context.get()
--- a/backend/app/db/session.py
+++ b/backend/app/db/session.py
@@ -1,3 +1,4 @@
+import logging
 from typing import AsyncGenerator

 from sqlalchemy import text
@@ -6,9 +7,20 @@ from sqlalchemy.orm import declarative_base

 from app.core.config import settings

+logger = logging.getLogger(__name__)
+
+DB_POOL_CONFIG = {
+    "pool_pre_ping": True,
+    "pool_recycle": 1800,
+    "pool_size": 10,
+    "max_overflow": 20,
+    "pool_timeout": 30,
+}
+
 engine = create_async_engine(
    settings.DATABASE_URL,
    echo=settings.DEBUG if hasattr(settings, "DEBUG") else False,
+    **DB_POOL_CONFIG,
 )

 async_session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
@@ -97,6 +109,16 @@ async def init_db():
    import app.models.system_setting  # noqa: F401
    import app.models.playground_session  # noqa: F401
    import app.models.playground_message  # noqa: F401
+    import app.models.system_log  # noqa: F401
+
+    logger.warning(
+        "Database pool settings active: pre_ping=%s recycle=%ss size=%s overflow=%s timeout=%ss",
+        DB_POOL_CONFIG["pool_pre_ping"],
+        DB_POOL_CONFIG["pool_recycle"],
+        DB_POOL_CONFIG["pool_size"],
+        DB_POOL_CONFIG["max_overflow"],
+        DB_POOL_CONFIG["pool_timeout"],
+    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -1,4 +1,5 @@
 from contextlib import asynccontextmanager
+from uuid import uuid4

 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
@@ -7,6 +8,7 @@ from starlette.middleware.base import BaseHTTPMiddleware
 from app.api.main import api_router
 from app.api.v1 import websocket
 from app.core.config import settings
+from app.core.request_context import set_request_id
 from app.core.websocket.broadcaster import broadcaster
 from app.db.session import init_db
 from app.services.scheduler import (
@@ -28,6 +30,18 @@ class WebSocketCORSMiddleware(BaseHTTPMiddleware):
        return await call_next(request)


+class RequestContextMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request, call_next):
+        request_id = request.headers.get("X-Request-ID") or uuid4().hex
+        set_request_id(request_id)
+        try:
+            response = await call_next(request)
+        finally:
+            set_request_id(None)
+        response.headers["X-Request-ID"] = request_id
+        return response
+
+
@asynccontextmanager
 async def lifespan(app: FastAPI):
    await init_db()
@@ -58,6 +72,7 @@ app.add_middleware(
    allow_headers=["*"],
 )

+app.add_middleware(RequestContextMiddleware)
 app.add_middleware(WebSocketCORSMiddleware)

 app.include_router(api_router, prefix="/api/v1")
--- a/backend/app/models/init.py
+++ b/backend/app/models/init.py
@@ -11,6 +11,7 @@ from app.models.bgp_observation import BGPObservation
 from app.models.system_setting import SystemSetting
 from app.models.playground_session import PlaygroundSession
 from app.models.playground_message import PlaygroundMessage
+from app.models.system_log import SystemLog, AuditLog

 __all__ = [
    "User",
@@ -26,4 +27,6 @@ __all__ = [
    "BGPAnomaly",
    "BGPIncident",
    "BGPObservation",
+    "SystemLog",
+    "AuditLog",
 ]
--- a/backend/app/models/system_log.py
+++ b/backend/app/models/system_log.py
@@ -0,0 +1,40 @@
+from sqlalchemy import JSON, Column, DateTime, Integer, String, Text
+from sqlalchemy.sql import func
+
+from app.db.session import Base
+
+
+class SystemLog(Base):
+    __tablename__ = "system_logs"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    occurred_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+    source = Column(String(50), nullable=False, index=True)
+    service = Column(String(50), nullable=True)
+    module = Column(String(120), nullable=True)
+    event = Column(String(160), nullable=True, index=True)
+    level = Column(String(20), nullable=False, index=True)
+    message = Column(Text, nullable=False)
+    request_id = Column(String(64), nullable=True, index=True)
+    trace_id = Column(String(64), nullable=True)
+    user_id = Column(Integer, nullable=True, index=True)
+    category = Column(String(80), nullable=True, index=True)
+    context = Column(JSON, nullable=False, default=dict)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+
+class AuditLog(Base):
+    __tablename__ = "audit_logs"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    occurred_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+    actor_id = Column(Integer, nullable=True, index=True)
+    actor_name = Column(String(255), nullable=True)
+    action = Column(String(120), nullable=False, index=True)
+    target_type = Column(String(80), nullable=True)
+    target_id = Column(String(120), nullable=True)
+    result = Column(String(40), nullable=True, index=True)
+    request_id = Column(String(64), nullable=True, index=True)
+    ip = Column(String(64), nullable=True)
+    details = Column(JSON, nullable=False, default=dict)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
--- a/backend/app/services/earth_news.py
+++ b/backend/app/services/earth_news.py
@@ -30,6 +30,14 @@ class RegionProfile:
    accent: str


+@dataclass(frozen=True)
+class RegionAnchor:
+    region: str
+    label: str
+    latitude: float
+    longitude: float
+
+
@dataclass(frozen=True)
 class NewsFeedSource:
    id: str
@@ -95,6 +103,39 @@ REGION_PROFILES: dict[str, RegionProfile] = {
    ),
 }

+REGION_ANCHORS: dict[str, RegionAnchor] = {
+    "americas": RegionAnchor(
+        region="americas",
+        label="美洲",
+        latitude=37.0902,
+        longitude=-95.7129,
+    ),
+    "europe": RegionAnchor(
+        region="europe",
+        label="欧洲",
+        latitude=50.1109,
+        longitude=8.6821,
+    ),
+    "middle-east-africa": RegionAnchor(
+        region="middle-east-africa",
+        label="中东与非洲",
+        latitude=25.2048,
+        longitude=55.2708,
+    ),
+    "asia-pacific": RegionAnchor(
+        region="asia-pacific",
+        label="亚太",
+        latitude=1.3521,
+        longitude=103.8198,
+    ),
+    "global": RegionAnchor(
+        region="global",
+        label="全球",
+        latitude=20.0,
+        longitude=0.0,
+    ),
+}
+

 def _google_news_feed(query: str, *, hl: str, gl: str, ceid: str) -> str:
    return (
@@ -213,6 +254,10 @@ def get_region_profile(region: str) -> RegionProfile:
    return REGION_PROFILES.get(region, REGION_PROFILES["global"])


+def get_region_anchor(region: str) -> RegionAnchor:
+    return REGION_ANCHORS.get(region, REGION_ANCHORS["global"])
+
+
 def get_sources_for_region(region: str) -> list[NewsFeedSource]:
    return sorted(
        [source for source in NEWS_FEED_SOURCES if source.region in {"global", region}],
@@ -342,6 +387,7 @@ def _serialize_sources(sources: list[NewsFeedSource]) -> list[dict[str, Any]]:

 def _serialize_item(item: ParsedNewsItem, *, active_region: str) -> dict[str, Any]:
    published_at = item.published_at
+    anchor = get_region_anchor(item.feed_region)
    return {
        "id": item.id,
        "title": item.title,
@@ -352,6 +398,10 @@ def _serialize_item(item: ParsedNewsItem, *, active_region: str) -> dict[str, An
        "region": item.feed_region,
        "homepage_url": item.homepage_url,
        "published_at": published_at.isoformat().replace("+00:00", "Z") if published_at else None,
+        "latitude": anchor.latitude,
+        "longitude": anchor.longitude,
+        "location_label": anchor.label,
+        "location_inferred": True,
        "is_focus_match": item.feed_region == active_region,
    }

--- a/backend/app/services/persistent_logs.py
+++ b/backend/app/services/persistent_logs.py
@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.core.request_context import get_request_id
+from app.db.session import async_session_factory
+from app.models.system_log import AuditLog, SystemLog
+
+logger = logging.getLogger(__name__)
+
+
+async def record_system_log(
+    *,
+    source: str,
+    level: str,
+    message: str,
+    service: str | None = None,
+    module: str | None = None,
+    event: str | None = None,
+    request_id: str | None = None,
+    trace_id: str | None = None,
+    user_id: int | None = None,
+    category: str | None = None,
+    context: dict[str, Any] | None = None,
+) -> None:
+    try:
+        async with async_session_factory() as session:
+            session.add(
+                SystemLog(
+                    source=source,
+                    service=service,
+                    module=module,
+                    event=event,
+                    level=level.lower(),
+                    message=message,
+                    request_id=request_id or get_request_id(),
+                    trace_id=trace_id,
+                    user_id=user_id,
+                    category=category,
+                    context=context or {},
+                )
+            )
+            await session.commit()
+    except Exception:
+        logger.exception("Failed to persist system log event=%s source=%s", event, source)
+
+
+async def record_audit_log(
+    *,
+    action: str,
+    actor_id: int | None = None,
+    actor_name: str | None = None,
+    target_type: str | None = None,
+    target_id: str | None = None,
+    result: str | None = None,
+    request_id: str | None = None,
+    ip: str | None = None,
+    details: dict[str, Any] | None = None,
+) -> None:
+    try:
+        async with async_session_factory() as session:
+            session.add(
+                AuditLog(
+                    actor_id=actor_id,
+                    actor_name=actor_name,
+                    action=action,
+                    target_type=target_type,
+                    target_id=target_id,
+                    result=result,
+                    request_id=request_id or get_request_id(),
+                    ip=ip,
+                    details=details or {},
+                )
+            )
+            await session.commit()
+    except Exception:
+        logger.exception("Failed to persist audit log action=%s", action)
--- a/backend/app/services/system_logs.py
+++ b/backend/app/services/system_logs.py
@@ -0,0 +1,522 @@
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import subprocess
+
+from collections import Counter, deque
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from app.core.security import redis_client
+
+DEFAULT_LOG_LINE_LIMIT = 200
+MAX_LOG_LINE_LIMIT = 1000
+BUFFER_LOG_LIMIT = 1000
+BUFFER_LOG_TTL_SECONDS = 7 * 24 * 60 * 60
+LOG_BUFFER_KEY_PREFIX = "planet:system_logs"
+
+LOG_LEVEL_ERROR = "error"
+LOG_LEVEL_WARNING = "warning"
+LOG_LEVEL_INFO = "info"
+LOG_LEVEL_DEBUG = "debug"
+LOG_LEVEL_ALL = "all"
+
+SUPPORTED_LOG_LEVELS = {
+    LOG_LEVEL_ALL,
+    LOG_LEVEL_ERROR,
+    LOG_LEVEL_WARNING,
+    LOG_LEVEL_INFO,
+    LOG_LEVEL_DEBUG,
+}
+
+LOG_LEVEL_ALIASES = {
+    "warn": LOG_LEVEL_WARNING,
+    "warning": LOG_LEVEL_WARNING,
+    "err": LOG_LEVEL_ERROR,
+    "error": LOG_LEVEL_ERROR,
+    "info": LOG_LEVEL_INFO,
+    "information": LOG_LEVEL_INFO,
+    "debug": LOG_LEVEL_DEBUG,
+    "trace": LOG_LEVEL_DEBUG,
+    "critical": LOG_LEVEL_ERROR,
+    "fatal": LOG_LEVEL_ERROR,
+}
+
+TIMESTAMP_FORMATS = (
+    "%Y-%m-%d %H:%M:%S.%f",
+    "%Y-%m-%d %H:%M:%S",
+    "%Y-%m-%dT%H:%M:%S.%f",
+    "%Y-%m-%dT%H:%M:%S",
+)
+
+LEVEL_PATTERNS = (
+    ("CRITICAL", LOG_LEVEL_ERROR),
+    ("FATAL", LOG_LEVEL_ERROR),
+    ("ERROR", LOG_LEVEL_ERROR),
+    ("WARNING", LOG_LEVEL_WARNING),
+    ("WARN", LOG_LEVEL_WARNING),
+    ("INFO", LOG_LEVEL_INFO),
+    ("DEBUG", LOG_LEVEL_DEBUG),
+    ("TRACE", LOG_LEVEL_DEBUG),
+)
+
+LEADING_LEVEL_PATTERN = re.compile(
+    r"^\s*(?:\[[^\]]+\]\s*)?(CRITICAL|FATAL|ERROR|WARNING|WARN|INFO|DEBUG|TRACE)\b[:\s-]*",
+    re.IGNORECASE,
+)
+EMBEDDED_LEVEL_PATTERN = re.compile(
+    r"\b(CRITICAL|FATAL|ERROR|WARNING|WARN|INFO|DEBUG|TRACE)\b",
+    re.IGNORECASE,
+)
+
+
+@dataclass(frozen=True)
+class LogSource:
+    source_id: str
+    name: str
+    kind: str
+    location: str
+    description: str
+    category: str
+    status: str = "ok"
+    buffer_key: str | None = None
+    container_name: str | None = None
+
+
+@dataclass
+class StructuredLogEntry:
+    timestamp: datetime | None
+    level: str | None
+    display_line: str
+    raw_line: str
+    search_text: str
+
+
+@dataclass
+class DailyLogMarker:
+    date_token: str
+    total: int
+    dominant_level: str
+
+
+LOG_SOURCES: dict[str, LogSource] = {
+    "backend": LogSource(
+        source_id="backend",
+        name="后端服务",
+        kind="file",
+        location="/tmp/planet_backend.log",
+        description="FastAPI 后端、调度器和采集任务共享日志。",
+        category="service",
+    ),
+    "frontend": LogSource(
+        source_id="frontend",
+        name="前端开发服务",
+        kind="file",
+        location="/tmp/planet_frontend.log",
+        description="控制台与 Earth 前端开发服务输出。",
+        category="service",
+    ),
+    "ai-provider": LogSource(
+        source_id="ai-provider",
+        name="AI Provider",
+        kind="docker",
+        location="docker://planet_aiprovider",
+        description="AI Provider 容器实时输出日志。",
+        category="service",
+        container_name="planet_aiprovider",
+    ),
+    "earth-client": LogSource(
+        source_id="earth-client",
+        name="Earth 浏览器端",
+        kind="buffer",
+        location="redis://planet:system_logs:earth-client",
+        description="Earth 浏览器端上报的运行时错误与关键业务日志。",
+        category="client",
+        buffer_key=f"{LOG_BUFFER_KEY_PREFIX}:earth-client",
+    ),
+}
+
+
+def normalize_log_level(level: str | None) -> str:
+    if level is None:
+        return LOG_LEVEL_ALL
+    normalized = str(level).strip().lower()
+    if normalized in {"", LOG_LEVEL_ALL}:
+        return LOG_LEVEL_ALL
+    return LOG_LEVEL_ALIASES.get(normalized, LOG_LEVEL_ALL)
+
+
+def normalize_log_levels(level: str | None = None, levels: str | None = None) -> tuple[str, ...]:
+    normalized_levels: list[str] = []
+    if levels:
+        for item in str(levels).split(","):
+            normalized = normalize_log_level(item)
+            if normalized != LOG_LEVEL_ALL and normalized not in normalized_levels:
+                normalized_levels.append(normalized)
+    normalized_level = normalize_log_level(level)
+    if normalized_level != LOG_LEVEL_ALL and normalized_level not in normalized_levels:
+        normalized_levels.append(normalized_level)
+    return tuple(normalized_levels)
+
+
+def get_source_status(source: LogSource) -> str:
+    if source.kind == "file":
+        path = Path(source.location)
+        if not path.exists():
+            return "missing"
+        return "ok" if path.stat().st_size > 0 else "empty"
+    if source.kind == "docker":
+        return "ok" if shutil.which("docker") else "docker_unavailable"
+    if source.kind == "buffer":
+        if not source.buffer_key:
+            return "source_unavailable"
+        try:
+            return "ok" if redis_client.llen(source.buffer_key) > 0 else "empty"
+        except Exception:
+            return "source_unavailable"
+    return "source_unavailable"
+
+
+def list_log_sources() -> list[dict[str, str]]:
+    items: list[dict[str, str]] = []
+    for source in LOG_SOURCES.values():
+        items.append(
+            {
+                "source_id": source.source_id,
+                "name": source.name,
+                "kind": source.kind,
+                "location": source.location,
+                "description": source.description,
+                "category": source.category,
+                "status": get_source_status(source),
+            }
+        )
+    return items
+
+
+def get_buffer_log_key(source_id: str) -> str:
+    return f"{LOG_BUFFER_KEY_PREFIX}:{source_id}"
+
+
+def append_buffer_log(
+    source_id: str,
+    *,
+    level: str,
+    message: str,
+    context: dict[str, Any] | None = None,
+) -> None:
+    payload = {
+        "timestamp": datetime.now(tz=UTC).isoformat(),
+        "level": normalize_log_level(level),
+        "message": message,
+        "context": context or {},
+    }
+    buffer_key = get_buffer_log_key(source_id)
+    redis_client.rpush(buffer_key, json.dumps(payload, ensure_ascii=False))
+    redis_client.ltrim(buffer_key, -BUFFER_LOG_LIMIT, -1)
+    redis_client.expire(buffer_key, BUFFER_LOG_TTL_SECONDS)
+
+
+def parse_timestamp(raw_value: str | None) -> datetime | None:
+    if not raw_value:
+        return None
+    candidate = str(raw_value).strip()
+    if not candidate:
+        return None
+    candidate = candidate.replace("Z", "+00:00")
+    try:
+        parsed = datetime.fromisoformat(candidate)
+        return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
+    except ValueError:
+        pass
+    for fmt in TIMESTAMP_FORMATS:
+        try:
+            return datetime.strptime(candidate, fmt).replace(tzinfo=UTC)
+        except ValueError:
+            continue
+    return None
+
+
+def parse_prefixed_timestamp(line: str) -> tuple[datetime | None, str]:
+    stripped = line.strip()
+    if not stripped:
+        return None, ""
+    for prefix_length in (35, 32, 29, 26, 23, 19):
+        if len(stripped) < prefix_length:
+            continue
+        prefix = stripped[:prefix_length]
+        timestamp = parse_timestamp(prefix)
+        if timestamp is not None:
+            return timestamp, stripped[prefix_length:].lstrip()
+    first_token = stripped.split(maxsplit=1)[0]
+    timestamp = parse_timestamp(first_token)
+    if timestamp is not None:
+        remainder = stripped[len(first_token):].lstrip()
+        return timestamp, remainder
+    return None, stripped
+
+
+def infer_log_level_from_text(text: str) -> str | None:
+    leading_match = LEADING_LEVEL_PATTERN.match(text)
+    if leading_match:
+        return normalize_log_level(leading_match.group(1))
+
+    embedded_match = EMBEDDED_LEVEL_PATTERN.search(text)
+    if embedded_match:
+        return normalize_log_level(embedded_match.group(1))
+
+    upper_text = text.upper()
+    for pattern, normalized in LEVEL_PATTERNS:
+        if f"{pattern}:" in upper_text or f"{pattern} " in upper_text:
+            return normalized
+    return None
+
+
+def build_display_line(timestamp: datetime | None, level: str | None, message: str) -> str:
+    message_part = message.strip() if message else ""
+    parts = []
+    if timestamp is not None:
+        parts.append(timestamp.astimezone(UTC).strftime("%Y-%m-%d %H:%M:%S"))
+    if level:
+        parts.append(level.upper())
+    if message_part:
+        parts.append(message_part)
+    return " ".join(parts).strip()
+
+
+def parse_text_log_entry(line: str) -> StructuredLogEntry:
+    timestamp, remainder = parse_prefixed_timestamp(line)
+    level = infer_log_level_from_text(remainder or line)
+    display_line = line.rstrip("\n")
+    return StructuredLogEntry(
+        timestamp=timestamp,
+        level=level,
+        display_line=display_line,
+        raw_line=display_line,
+        search_text=display_line.lower(),
+    )
+
+
+def build_buffer_entry(payload: dict[str, Any]) -> StructuredLogEntry:
+    timestamp = parse_timestamp(str(payload.get("timestamp", "")).strip())
+    level = normalize_log_level(payload.get("level"))
+    if level == LOG_LEVEL_ALL:
+        level = None
+    message = str(payload.get("message", "")).strip()
+    context = payload.get("context")
+    context_map = context if isinstance(context, dict) else {}
+    context_fragments = []
+    for key in ("category", "module", "url", "detail"):
+        value = str(context_map.get(key, "")).strip()
+        if value:
+            context_fragments.append(f"{key}={value}")
+    message_with_context = " | ".join([message, *context_fragments]) if context_fragments else message
+    display_line = build_display_line(timestamp, level, message_with_context)
+    search_text = " ".join(
+        [
+            message,
+            json.dumps(context_map, ensure_ascii=False, sort_keys=True),
+            display_line,
+        ]
+    ).lower()
+    return StructuredLogEntry(
+        timestamp=timestamp,
+        level=level,
+        display_line=display_line,
+        raw_line=json.dumps(payload, ensure_ascii=False, sort_keys=True),
+        search_text=search_text,
+    )
+
+
+def read_file_entries(source: LogSource, scan_limit: int) -> list[StructuredLogEntry]:
+    path = Path(source.location)
+    if not path.exists():
+        return []
+    with path.open("r", encoding="utf-8", errors="replace") as handle:
+        recent_lines = deque(handle, maxlen=scan_limit)
+    return [parse_text_log_entry(line) for line in recent_lines if line.strip()]
+
+
+def read_docker_entries(source: LogSource, scan_limit: int) -> list[StructuredLogEntry]:
+    if not shutil.which("docker") or not source.container_name:
+        return []
+    try:
+        completed = subprocess.run(
+            [
+                "docker",
+                "logs",
+                "--timestamps",
+                "--tail",
+                str(scan_limit),
+                source.container_name,
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError:
+        return []
+    if completed.returncode != 0:
+        return []
+    return [
+        parse_text_log_entry(line)
+        for line in completed.stdout.splitlines()
+        if line.strip()
+    ]
+
+
+def read_buffer_entries(source: LogSource, scan_limit: int) -> list[StructuredLogEntry]:
+    if not source.buffer_key:
+        return []
+    try:
+        raw_items = redis_client.lrange(source.buffer_key, -scan_limit, -1)
+    except Exception:
+        return []
+    entries: list[StructuredLogEntry] = []
+    for raw_item in raw_items:
+        try:
+            payload = json.loads(raw_item)
+        except json.JSONDecodeError:
+            entries.append(parse_text_log_entry(str(raw_item)))
+            continue
+        if isinstance(payload, dict):
+            entries.append(build_buffer_entry(payload))
+        else:
+            entries.append(parse_text_log_entry(str(raw_item)))
+    return entries
+
+
+def read_source_entries(source: LogSource, scan_limit: int) -> list[StructuredLogEntry]:
+    if source.kind == "file":
+        return read_file_entries(source, scan_limit)
+    if source.kind == "docker":
+        return read_docker_entries(source, scan_limit)
+    if source.kind == "buffer":
+        return read_buffer_entries(source, scan_limit)
+    return []
+
+
+def matches_levels(entry: StructuredLogEntry, selected_levels: tuple[str, ...]) -> bool:
+    if not selected_levels:
+        return True
+    return entry.level in selected_levels
+
+
+def matches_date_range(
+    entry: StructuredLogEntry,
+    start_date: str | None,
+    end_date: str | None,
+) -> bool:
+    if not start_date and not end_date:
+        return True
+    if entry.timestamp is None:
+        return False
+    date_token = entry.timestamp.astimezone(UTC).date().isoformat()
+    if start_date and date_token < start_date:
+        return False
+    if end_date and date_token > end_date:
+        return False
+    return True
+
+
+def matches_search(entry: StructuredLogEntry, search: str | None) -> bool:
+    if search is None:
+        return True
+    query = search.strip().lower()
+    if not query:
+        return True
+    return query in entry.search_text
+
+
+def build_daily_log_markers(entries: list[StructuredLogEntry]) -> list[dict[str, Any]]:
+    grouped: dict[str, list[StructuredLogEntry]] = {}
+    for entry in entries:
+        if entry.timestamp is None:
+            continue
+        date_token = entry.timestamp.astimezone(UTC).date().isoformat()
+        grouped.setdefault(date_token, []).append(entry)
+
+    markers: list[DailyLogMarker] = []
+    for date_token, group in sorted(grouped.items()):
+        level_counts = Counter(
+            entry.level
+            for entry in group
+            if entry.level in SUPPORTED_LOG_LEVELS and entry.level != LOG_LEVEL_ALL
+        )
+        dominant_level = LOG_LEVEL_INFO
+        if level_counts:
+            dominant_level = sorted(
+                level_counts.items(),
+                key=lambda item: (
+                    -item[1],
+                    ("error", "warning", "info", "debug").index(item[0]),
+                ),
+            )[0][0]
+        markers.append(
+            DailyLogMarker(
+                date_token=date_token,
+                total=len(group),
+                dominant_level=dominant_level,
+            )
+        )
+    return [marker.__dict__ for marker in markers]
+
+
+def read_log_snapshot(
+    source_id: str,
+    limit: int,
+    *,
+    level: str = LOG_LEVEL_ALL,
+    levels: str | None = None,
+    start_date: str | None = None,
+    end_date: str | None = None,
+    search: str | None = None,
+) -> dict[str, Any] | None:
+    source = LOG_SOURCES.get(source_id)
+    if source is None:
+        return None
+
+    selected_levels = normalize_log_levels(level, levels)
+    search_query = (search or "").strip()
+    scan_limit = max(min(MAX_LOG_LINE_LIMIT * 5, 5000), limit * 5, BUFFER_LOG_LIMIT if source.kind == "buffer" else 1000)
+    all_entries = read_source_entries(source, scan_limit)
+    marker_entries = [
+        entry
+        for entry in all_entries
+        if matches_levels(entry, selected_levels) and matches_search(entry, search_query)
+    ]
+    filtered_entries = [
+        entry
+        for entry in marker_entries
+        if matches_date_range(entry, start_date, end_date)
+    ]
+    visible_entries = filtered_entries[-limit:]
+
+    compatibility_level = selected_levels[0] if len(selected_levels) == 1 else LOG_LEVEL_ALL
+    return {
+        "source_id": source.source_id,
+        "name": source.name,
+        "kind": source.kind,
+        "location": source.location,
+        "description": source.description,
+        "category": source.category,
+        "status": get_source_status(source),
+        "level": compatibility_level,
+        "selected_levels": list(selected_levels),
+        "search_query": search_query,
+        "available_levels": [
+            LOG_LEVEL_ALL,
+            LOG_LEVEL_ERROR,
+            LOG_LEVEL_WARNING,
+            LOG_LEVEL_INFO,
+            LOG_LEVEL_DEBUG,
+        ],
+        "daily_markers": build_daily_log_markers(marker_entries),
+        "line_limit": limit,
+        "line_count": len(visible_entries),
+        "lines": [entry.display_line for entry in visible_entries],
+    }
--- a/backend/tests/test_api.py
+++ b/backend/tests/test_api.py
@@ -35,6 +35,7 @@ async def test_health_check():
    data = response.json()
    assert data["status"] == "healthy"
    assert "version" in data
+    assert response.headers["x-request-id"]


@pytest.mark.asyncio
@@ -161,6 +162,345 @@ async def test_alerts_endpoint_with_auth(auth_headers):
        app.dependency_overrides.clear()


+@pytest.mark.asyncio
+async def test_system_log_sources_requires_super_admin(auth_headers):
+    def override_get_current_user():
+        return User(
+            id=1,
+            username="testuser",
+            email="test@example.com",
+            password_hash="hashed",
+            role="admin",
+            is_active=True,
+        )
+
+    app.dependency_overrides = {
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: override_get_current_user,
+    }
+    transport = ASGITransport(app=app)
+    try:
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            response = await client.get("/api/v1/system/logs/sources", headers=auth_headers)
+        assert response.status_code == 403
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_system_log_sources_with_super_admin(auth_headers):
+    def override_get_current_user():
+        return User(
+            id=1,
+            username="root",
+            email="root@example.com",
+            password_hash="hashed",
+            role="super_admin",
+            is_active=True,
+        )
+
+    app.dependency_overrides = {
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: override_get_current_user,
+    }
+    transport = ASGITransport(app=app)
+    try:
+        with patch(
+            "app.api.v1.system_control.list_log_sources",
+            return_value=[
+                {
+                    "source_id": "backend",
+                    "name": "后端服务",
+                    "kind": "file",
+                    "location": "/tmp/planet_backend.log",
+                    "description": "FastAPI 后端、调度器和采集任务共享日志。",
+                    "category": "service",
+                    "status": "ok",
+                }
+            ],
+        ):
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                response = await client.get("/api/v1/system/logs/sources", headers=auth_headers)
+        assert response.status_code == 200
+        data = response.json()
+        assert data["items"][0]["source_id"] == "backend"
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_system_log_snapshot_with_super_admin(auth_headers):
+    def override_get_current_user():
+        return User(
+            id=1,
+            username="root",
+            email="root@example.com",
+            password_hash="hashed",
+            role="super_admin",
+            is_active=True,
+        )
+
+    app.dependency_overrides = {
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: override_get_current_user,
+    }
+    transport = ASGITransport(app=app)
+    try:
+        with patch(
+            "app.api.v1.system_control.read_log_snapshot",
+            return_value={
+                "source_id": "backend",
+                "name": "后端服务",
+                "kind": "file",
+                "location": "/tmp/planet_backend.log",
+                "description": "FastAPI 后端、调度器和采集任务共享日志。",
+                "category": "service",
+                "status": "ok",
+                "level": "all",
+                "selected_levels": [],
+                "search_query": "",
+                "available_levels": ["all", "error", "warning", "info", "debug"],
+                "daily_markers": [],
+                "line_limit": 50,
+                "line_count": 2,
+                "lines": ["line 1", "line 2"],
+            },
+        ):
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                response = await client.get("/api/v1/system/logs/backend?limit=50", headers=auth_headers)
+        assert response.status_code == 200
+        data = response.json()
+        assert data["source_id"] == "backend"
+        assert data["line_count"] == 2
+        assert data["lines"] == ["line 1", "line 2"]
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_system_log_snapshot_supports_level_filter(auth_headers):
+    def override_get_current_user():
+        return User(
+            id=1,
+            username="root",
+            email="root@example.com",
+            password_hash="hashed",
+            role="super_admin",
+            is_active=True,
+        )
+
+    app.dependency_overrides = {
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: override_get_current_user,
+    }
+    transport = ASGITransport(app=app)
+    try:
+        with patch(
+            "app.api.v1.system_control.read_log_snapshot",
+            return_value={
+                "source_id": "backend",
+                "name": "后端服务",
+                "kind": "file",
+                "location": "/tmp/planet_backend.log",
+                "description": "FastAPI 后端、调度器和采集任务共享日志。",
+                "category": "service",
+                "status": "ok",
+                "level": "error",
+                "selected_levels": ["error"],
+                "search_query": "",
+                "available_levels": ["all", "error", "warning", "info", "debug"],
+                "daily_markers": [],
+                "line_limit": 50,
+                "line_count": 1,
+                "lines": ["ERROR: failed"],
+            },
+        ):
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                response = await client.get("/api/v1/system/logs/backend?limit=50&level=error", headers=auth_headers)
+        assert response.status_code == 200
+        data = response.json()
+        assert data["level"] == "error"
+        assert data["lines"] == ["ERROR: failed"]
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_system_log_snapshot_supports_date_range_filter(auth_headers):
+    def override_get_current_user():
+        return User(
+            id=1,
+            username="root",
+            email="root@example.com",
+            password_hash="hashed",
+            role="super_admin",
+            is_active=True,
+        )
+
+    app.dependency_overrides = {
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: override_get_current_user,
+    }
+    transport = ASGITransport(app=app)
+    try:
+        with patch(
+            "app.api.v1.system_control.read_log_snapshot",
+            return_value={
+                "source_id": "backend",
+                "name": "后端服务",
+                "kind": "file",
+                "location": "/tmp/planet_backend.log",
+                "description": "FastAPI 后端、调度器和采集任务共享日志。",
+                "category": "service",
+                "status": "ok",
+                "level": "all",
+                "selected_levels": [],
+                "search_query": "",
+                "available_levels": ["all", "error", "warning", "info", "debug"],
+                "daily_markers": [],
+                "line_limit": 50,
+                "line_count": 1,
+                "lines": ["2026-04-23 INFO: service started"],
+            },
+        ) as mock_read_log_snapshot:
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                response = await client.get(
+                    "/api/v1/system/logs/backend?limit=50&start_date=2026-04-20&end_date=2026-04-23",
+                    headers=auth_headers,
+                )
+        assert response.status_code == 200
+        mock_read_log_snapshot.assert_called_once_with(
+            "backend",
+            50,
+            level="all",
+            levels=None,
+            start_date="2026-04-20",
+            end_date="2026-04-23",
+            search=None,
+        )
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_system_log_snapshot_supports_levels_and_search_filter(auth_headers):
+    def override_get_current_user():
+        return User(
+            id=1,
+            username="root",
+            email="root@example.com",
+            password_hash="hashed",
+            role="super_admin",
+            is_active=True,
+        )
+
+    app.dependency_overrides = {
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: override_get_current_user,
+    }
+    transport = ASGITransport(app=app)
+    try:
+        with patch(
+            "app.api.v1.system_control.read_log_snapshot",
+            return_value={
+                "source_id": "backend",
+                "name": "后端服务",
+                "kind": "file",
+                "location": "/tmp/planet_backend.log",
+                "description": "FastAPI 后端、调度器和采集任务共享日志。",
+                "category": "service",
+                "status": "ok",
+                "level": "all",
+                "selected_levels": ["error", "warning"],
+                "search_query": "timeout",
+                "available_levels": ["all", "error", "warning", "info", "debug"],
+                "daily_markers": [],
+                "line_limit": 50,
+                "line_count": 1,
+                "lines": ["2026-04-23 10:00:00 ERROR timeout"],
+            },
+        ) as mock_read_log_snapshot:
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                response = await client.get(
+                    "/api/v1/system/logs/backend?limit=50&levels=error,warning&search=timeout",
+                    headers=auth_headers,
+                )
+        assert response.status_code == 200
+        mock_read_log_snapshot.assert_called_once_with(
+            "backend",
+            50,
+            level="all",
+            levels="error,warning",
+            start_date=None,
+            end_date=None,
+            search="timeout",
+        )
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_system_log_snapshot_rejects_invalid_date_range(auth_headers):
+    def override_get_current_user():
+        return User(
+            id=1,
+            username="root",
+            email="root@example.com",
+            password_hash="hashed",
+            role="super_admin",
+            is_active=True,
+        )
+
+    app.dependency_overrides = {
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: override_get_current_user,
+    }
+    transport = ASGITransport(app=app)
+    try:
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            response = await client.get(
+                "/api/v1/system/logs/backend?start_date=2026-04-31",
+                headers=auth_headers,
+            )
+        assert response.status_code == 400
+        assert "start_date must be in YYYY-MM-DD format" in response.json()["detail"]
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_ingest_earth_client_log_accepts_public_events():
+    transport = ASGITransport(app=app)
+    try:
+        with patch("app.api.v1.system_control.append_buffer_log") as mock_append_buffer_log:
+            with patch("app.api.v1.system_control.record_system_log", new_callable=AsyncMock) as mock_record_system_log:
+                async with AsyncClient(transport=transport, base_url="http://test") as client:
+                    response = await client.post(
+                        "/api/v1/system/logs/earth-client",
+                        json={
+                            "level": "error",
+                            "message": "登陆点加载失败: 登陆点接口返回 HTTP 500",
+                            "category": "startup-load",
+                            "module": "layer-startup",
+                        },
+                    )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["accepted"] is True
+        assert data["source_id"] == "earth-client"
+        mock_append_buffer_log.assert_called_once()
+        mock_record_system_log.assert_awaited_once()
+        persisted_kwargs = mock_record_system_log.await_args.kwargs
+        assert persisted_kwargs["source"] == "earth-client"
+        assert persisted_kwargs["event"] == "earth.client.runtime_log"
+        assert persisted_kwargs["category"] == "startup-load"
+        assert persisted_kwargs["level"] == "error"
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_request_id_header_is_echoed_when_provided():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get("/health", headers={"X-Request-ID": "planet-test-request"})
+    assert response.status_code == 200
+    assert response.headers["x-request-id"] == "planet-test-request"
+
+
@pytest.mark.asyncio
 async def test_invalid_token():
    """Test that invalid token is rejected"""
@@ -263,6 +603,8 @@ async def test_ai_situational_analysis_returns_503_when_disabled(auth_headers):
        assert "content_blocks" in data
        assert "text_blocks" in data
        assert "thinking_blocks" in data
+    finally:
+        app.dependency_overrides.clear()


@pytest.mark.asyncio
@@ -382,8 +724,6 @@ async def test_save_playground_session_with_auth(auth_headers):
        assert data["state"]["objective"] == "测试目标"
    finally:
        app.dependency_overrides.clear()
-    finally:
-        app.dependency_overrides.clear()


@pytest.mark.asyncio
--- a/backend/tests/test_earth_news.py
+++ b/backend/tests/test_earth_news.py
@@ -0,0 +1,49 @@
+from datetime import UTC, datetime
+
+from app.services.earth_news import ParsedNewsItem, _serialize_item
+
+
+def test_serialize_item_includes_region_anchor_for_cruise():
+    item = ParsedNewsItem(
+        id="google-apac:test",
+        title="Example APAC story",
+        summary="Example summary",
+        url="https://example.com/story",
+        source="Example Source",
+        feed_name="Global Monitor / APAC",
+        feed_region="asia-pacific",
+        homepage_url="https://example.com",
+        published_at=datetime(2026, 4, 23, 2, 30, tzinfo=UTC),
+    )
+
+    payload = _serialize_item(item, active_region="asia-pacific")
+
+    assert payload["latitude"] == 1.3521
+    assert payload["longitude"] == 103.8198
+    assert payload["location_label"] == "亚太"
+    assert payload["location_inferred"] is True
+    assert payload["is_focus_match"] is True
+    assert payload["published_at"] == "2026-04-23T02:30:00Z"
+
+
+def test_serialize_item_falls_back_to_global_anchor():
+    item = ParsedNewsItem(
+        id="custom:test",
+        title="Fallback story",
+        summary="Fallback summary",
+        url="https://example.com/fallback",
+        source="Fallback Source",
+        feed_name="Fallback Feed",
+        feed_region="unknown-region",
+        homepage_url="https://example.com",
+        published_at=None,
+    )
+
+    payload = _serialize_item(item, active_region="americas")
+
+    assert payload["latitude"] == 20.0
+    assert payload["longitude"] == 0.0
+    assert payload["location_label"] == "全球"
+    assert payload["location_inferred"] is True
+    assert payload["is_focus_match"] is False
+    assert payload["published_at"] is None
--- a/backend/tests/test_system_logs.py
+++ b/backend/tests/test_system_logs.py
@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+import json
+
+from pathlib import Path
+
+from app.services import system_logs
+
+
+class FakeRedis:
+    def __init__(self) -> None:
+        self.store: dict[str, list[str]] = {}
+
+    def rpush(self, key: str, value: str) -> None:
+        self.store.setdefault(key, []).append(value)
+
+    def ltrim(self, key: str, start: int, end: int) -> None:
+        items = self.store.get(key, [])
+        normalized_end = None if end == -1 else end + 1
+        self.store[key] = items[start:normalized_end]
+
+    def expire(self, key: str, seconds: int) -> None:
+        return None
+
+    def lrange(self, key: str, start: int, end: int) -> list[str]:
+        items = self.store.get(key, [])
+        normalized_end = None if end == -1 else end + 1
+        return items[start:normalized_end]
+
+    def llen(self, key: str) -> int:
+        return len(self.store.get(key, []))
+
+
+def test_read_log_snapshot_uses_structured_buffer_timestamp_level_and_search(monkeypatch):
+    fake_redis = FakeRedis()
+    monkeypatch.setattr(system_logs, "redis_client", fake_redis)
+    monkeypatch.setattr(
+        system_logs,
+        "LOG_SOURCES",
+        {
+            "earth-client": system_logs.LogSource(
+                source_id="earth-client",
+                name="Earth 浏览器端",
+                kind="buffer",
+                location="redis://planet:system_logs:earth-client",
+                description="Earth 浏览器端上报日志",
+                category="client",
+                buffer_key=system_logs.get_buffer_log_key("earth-client"),
+            )
+        },
+    )
+
+    fake_redis.rpush(
+        system_logs.get_buffer_log_key("earth-client"),
+        json.dumps(
+            {
+                "timestamp": "2026-04-22T10:15:30Z",
+                "level": "warning",
+                "message": "news feed degraded",
+                "context": {"module": "news", "detail": "timeout"},
+            },
+            ensure_ascii=False,
+        ),
+    )
+    fake_redis.rpush(
+        system_logs.get_buffer_log_key("earth-client"),
+        json.dumps(
+            {
+                "timestamp": "2026-04-23T06:01:00Z",
+                "level": "error",
+                "message": "landing points failed",
+                "context": {"module": "layer-startup", "detail": "http 500"},
+            },
+            ensure_ascii=False,
+        ),
+    )
+
+    snapshot = system_logs.read_log_snapshot(
+        "earth-client",
+        50,
+        levels="error,warning",
+        start_date="2026-04-23",
+        end_date="2026-04-23",
+        search="landing",
+    )
+
+    assert snapshot is not None
+    assert snapshot["selected_levels"] == ["error", "warning"]
+    assert snapshot["search_query"] == "landing"
+    assert snapshot["line_count"] == 1
+    assert snapshot["lines"][0].startswith("2026-04-23 06:01:00 ERROR landing points failed")
+    assert snapshot["daily_markers"] == [
+        {"date_token": "2026-04-23", "total": 1, "dominant_level": "error"}
+    ]
+
+
+def test_read_log_snapshot_parses_file_timestamp_and_builds_markers(tmp_path: Path, monkeypatch):
+    log_path = tmp_path / "backend.log"
+    log_path.write_text(
+        "\n".join(
+            [
+                "2026-04-22 08:00:00 INFO service booted",
+                "2026-04-23 09:15:00 WARNING disk pressure detected",
+                "2026-04-23 09:16:00 ERROR sync failed",
+                "2026-04-24 10:00:00 DEBUG collector trace",
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(
+        system_logs,
+        "LOG_SOURCES",
+        {
+            "backend": system_logs.LogSource(
+                source_id="backend",
+                name="后端服务",
+                kind="file",
+                location=str(log_path),
+                description="测试文件日志",
+                category="service",
+            )
+        },
+    )
+
+    snapshot = system_logs.read_log_snapshot(
+        "backend",
+        50,
+        levels="warning,error",
+        search="failed",
+    )
+
+    assert snapshot is not None
+    assert snapshot["line_count"] == 1
+    assert snapshot["lines"] == ["2026-04-23 09:16:00 ERROR sync failed"]
+    assert snapshot["daily_markers"] == [
+        {"date_token": "2026-04-23", "total": 1, "dominant_level": "error"}
+    ]
+    assert snapshot["status"] == "ok"
+
+
+def test_append_buffer_log_persists_normalized_level(monkeypatch):
+    fake_redis = FakeRedis()
+    monkeypatch.setattr(system_logs, "redis_client", fake_redis)
+
+    system_logs.append_buffer_log(
+        "earth-client",
+        level="warn",
+        message="feed delayed",
+        context={"module": "news"},
+    )
+
+    stored_items = fake_redis.lrange(system_logs.get_buffer_log_key("earth-client"), 0, -1)
+    payload = json.loads(stored_items[0])
+    assert payload["level"] == "warning"
+    assert payload["message"] == "feed delayed"
+
+
+def test_infer_log_level_prefers_leading_prefix_over_query_string():
+    line = 'INFO: 127.0.0.1 - "GET /api/v1/system/logs/backend?limit=200&level=error&levels=error HTTP/1.1" 200 OK'
+
+    entry = system_logs.parse_text_log_entry(line)
+
+    assert entry.level == "info"