Byted Sol Stability Grafana Dashboard Design

MIT License

Copyright (c) 2026 ByteDance

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "byted-sol-stablity-grafana-dashboard-design"
version = "0.1.0"
description = "Generate Dashboard IA, panel specs, and Grafana JSON from SLI + architecture + code inputs"
readme = "SKILL.md"
requires-python = ">=3.10"
dependencies = ["pyyaml>=6.0"]

[project.scripts]
byted-sol-stablity-grafana-dashboard-design = "grafana_dashboard_design_skill.cli:main"

[tool.setuptools.packages.find]
where = ["src"]

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from .pipeline import run_pipeline

__all__ = ["run_pipeline"]

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from .arch_adapter import load_arch_model
from .metric_adapter import discover_code_metric_hints
from .sli_adapter import load_sli_model

__all__ = ["load_sli_model", "load_arch_model", "discover_code_metric_hints"]

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from pathlib import Path
import json
from typing import Any, Dict, List


REQUIRED_FILES = [
    "topology-model.json",
    "core-links.md",
    "dependency-risk.md",
    "observability-gaps.md",
    "evidence-index.json",
]


def _read_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def _read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def load_arch_model(path: str) -> Dict[str, Any]:
    root = Path(path)
    if not root.exists() or not root.is_dir():
        raise ValueError(f"arch model directory not found: {root}")

    missing: List[str] = []
    for filename in REQUIRED_FILES:
        if not (root / filename).exists():
            missing.append(filename)
    if missing:
        raise ValueError(f"arch model missing required files: {', '.join(missing)}")

    return {
        "topology_model": _read_json(root / "topology-model.json"),
        "core_links_md": _read_text(root / "core-links.md"),
        "dependency_risk_md": _read_text(root / "dependency-risk.md"),
        "observability_gaps_md": _read_text(root / "observability-gaps.md"),
        "evidence_index": _read_json(root / "evidence-index.json"),
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from pathlib import Path
import re
from typing import List, Set


METRIC_PATTERN = re.compile(
    r"\b([a-zA-Z_:][a-zA-Z0-9_:]*?(?:_total|_count|_error_rate|_success_rate|_latency_ms|_duration_ms|_p99|_p95))\b"
)
SUPPORTED_SUFFIXES = {".py", ".ts", ".js", ".go", ".java", ".yml", ".yaml", ".json", ".md"}
MAX_SCAN_FILES = 600


def discover_code_metric_hints(repo_path: str) -> List[str]:
    root = Path(repo_path)
    if not root.exists() or not root.is_dir():
        raise ValueError(f"repo path not found: {root}")

    collected: Set[str] = set()
    scanned = 0
    for path in root.rglob("*"):
        if scanned >= MAX_SCAN_FILES:
            break
        if not path.is_file() or path.suffix.lower() not in SUPPORTED_SUFFIXES:
            continue
        scanned += 1
        try:
            text = path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            continue
        for match in METRIC_PATTERN.findall(text):
            collected.add(match)

    return sorted(collected)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from pathlib import Path
import json
from typing import Any, Dict


def _load_json(path: Path) -> Dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def _pick_sli_json_file(root: Path) -> Path:
    preferred = [
        "sli-spec.json",
        "sli-spec.v2.all.json",
        "sli-spec.all.json",
    ]
    for name in preferred:
        candidate = root / name
        if candidate.exists() and candidate.is_file():
            return candidate
    json_files = sorted(path for path in root.glob("*.json") if path.is_file())
    if not json_files:
        raise ValueError(f"no json file found in sli model directory: {root}")
    return json_files[0]


def load_sli_model(path: str) -> Dict[str, Any]:
    target = Path(path)
    if not target.exists():
        raise ValueError(f"sli model path not found: {target}")
    if target.is_dir():
        target = _pick_sli_json_file(target)
    return _load_json(target)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import List

from .pipeline import PipelineOptions, run_pipeline


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Generate Dashboard IA, panel specs, and Grafana JSON")
    parser.add_argument("--sli-model", required=True, help="Path to SLI model JSON or directory")
    parser.add_argument("--arch-model", required=True, help="Path to architecture model directory")
    parser.add_argument("--repo", required=True, help="Path to source repository")
    parser.add_argument("--out-dir", default="output", help="Output base directory")
    parser.add_argument("--dashboard-title", help="Optional dashboard title override")
    parser.add_argument("--focus-service", help="Optional focus service")
    parser.add_argument("--offline", action="store_true", help="Use local inputs only")
    return parser


def run_cli(argv: List[str] | None = None) -> int:
    args = _parser().parse_args(argv)

    sli_model = Path(args.sli_model)
    arch_model = Path(args.arch_model)
    repo = Path(args.repo)

    if not sli_model.exists():
        raise SystemExit(f"sli model path not found: {sli_model}")
    if not arch_model.exists() or not arch_model.is_dir():
        raise SystemExit(f"arch model directory not found: {arch_model}")
    if not repo.exists() or not repo.is_dir():
        raise SystemExit(f"repo path not found: {repo}")

    result = run_pipeline(
        sli_model=str(sli_model),
        arch_model=str(arch_model),
        repo=str(repo),
        options=PipelineOptions(
            out_dir=args.out_dir,
            dashboard_title=args.dashboard_title,
            focus_service=args.focus_service,
            offline=args.offline,
        ),
    )

    print(json.dumps(result.to_dict(), ensure_ascii=False, indent=2))
    return 0


def main() -> None:
    raise SystemExit(run_cli())


if __name__ == "__main__":
    main()

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Dict, List, Tuple

from .models import DashboardIA, GrafanaDashboardDoc, GrafanaPanelDTO, PanelSpec


LAYER_ORDER = ["api", "gateway", "conversation", "queue", "agent", "llm", "cron"]


def _slug(text: str) -> str:
    result = []
    for ch in text.strip().lower():
        result.append(ch if ch.isalnum() else "-")
    slug = "".join(result)
    while "--" in slug:
        slug = slug.replace("--", "-")
    return slug.strip("-") or "dashboard"


def _zone_prefix(zone_type: str) -> str:
    if zone_type == "health":
        return "[HEALTH]"
    if zone_type == "diagnostic":
        return "[DIAGNOSTIC]"
    return "[EVENT]"


def _threshold_steps(spec: PanelSpec) -> List[Dict[str, float | str | None]]:
    steps: List[Dict[str, float | str | None]] = [{"color": "green", "value": None}]
    for rule in spec.thresholds:
        steps.append({"color": rule.color, "value": rule.value})
    return steps


def _panel_size(spec: PanelSpec) -> Tuple[int, int]:
    if spec.chart_type == "stat":
        return 4, 5
    if spec.chart_type == "table":
        return 8, 6
    if spec.chart_type == "state-timeline":
        return 12, 6
    return 12, 6


def _layout_overview_row1(section_panels: List[PanelSpec], y_start: int) -> Tuple[Dict[str, Dict[str, int]], int]:
    positions: Dict[str, Dict[str, int]] = {}
    by_id = {panel.panel_id: panel for panel in section_panels}

    stat_ids = [
        "overview-kpi-availability",
        "overview-kpi-slo-attainment",
        "overview-kpi-error-budget",
        "overview-kpi-active-alerts",
    ]
    x = 0
    for panel_id in stat_ids:
        panel = by_id.get(panel_id)
        if panel is None:
            continue
        positions[panel.panel_id] = {"x": x, "y": y_start, "w": 4, "h": 5}
        x += 4

    oncall = by_id.get("overview-kpi-oncall-topn")
    if oncall is not None:
        positions[oncall.panel_id] = {"x": 16, "y": y_start, "w": 8, "h": 5}

    return positions, y_start + 6


def _layout_overview_row2(section_panels: List[PanelSpec], y_start: int) -> Tuple[Dict[str, Dict[str, int]], int]:
    positions: Dict[str, Dict[str, int]] = {}
    by_id = {panel.panel_id: panel for panel in section_panels}

    scorecards = by_id.get("overview-sli-scorecards")
    trend = by_id.get("overview-sli-trend")
    topn_ids = [
        "overview-topn-impact-tenant",
        "overview-topn-impact-model",
        "overview-topn-impact-channel",
        "overview-topn-impact-agent",
    ]

    if scorecards is not None:
        positions[scorecards.panel_id] = {"x": 0, "y": y_start, "w": 6, "h": 12}
    if trend is not None:
        positions[trend.panel_id] = {"x": 6, "y": y_start, "w": 10, "h": 12}

    topn_y = y_start
    for topn_id in topn_ids:
        panel = by_id.get(topn_id)
        if panel is None:
            continue
        positions[panel.panel_id] = {"x": 16, "y": topn_y, "w": 8, "h": 3}
        topn_y += 3

    return positions, y_start + 13


def _layout_overview_row3(section_panels: List[PanelSpec], y_start: int) -> Tuple[Dict[str, Dict[str, int]], int]:
    positions: Dict[str, Dict[str, int]] = {}
    by_id = {panel.panel_id: panel for panel in section_panels}

    panel_ids = [
        "overview-events-alert-stream",
        "overview-events-reliability",
        "overview-events-change",
    ]
    x = 0
    for panel_id in panel_ids:
        panel = by_id.get(panel_id)
        if panel is None:
            continue
        positions[panel.panel_id] = {"x": x, "y": y_start, "w": 8, "h": 7}
        x += 8

    return positions, y_start + 8


def _layer_from_panel_id(panel_id: str) -> str:
    if not panel_id.startswith("service-layers-"):
        return ""
    parts = panel_id.split("-")
    if len(parts) < 4:
        return ""
    return parts[2]


def _layout_service_layers_health(
    section_panels: List[PanelSpec],
    y_start: int,
) -> Tuple[Dict[str, Dict[str, int]], int]:
    positions: Dict[str, Dict[str, int]] = {}
    grouped: Dict[str, Dict[str, PanelSpec]] = {key: {} for key in LAYER_ORDER}

    for panel in section_panels:
        layer = _layer_from_panel_id(panel.panel_id)
        if layer in grouped:
            if panel.panel_id.endswith("-scorecards"):
                grouped[layer]["scorecards"] = panel
            elif panel.panel_id.endswith("-availability"):
                grouped[layer]["availability"] = panel

    y = y_start
    for layer in LAYER_ORDER:
        scorecards = grouped[layer].get("scorecards")
        availability = grouped[layer].get("availability")
        if scorecards is not None:
            positions[scorecards.panel_id] = {"x": 0, "y": y, "w": 6, "h": 5}
        if availability is not None:
            positions[availability.panel_id] = {"x": 6, "y": y, "w": 18, "h": 5}
        y += 6

    return positions, y


def _layout_service_layers_diagnostic(
    section_panels: List[PanelSpec],
    y_start: int,
) -> Tuple[Dict[str, Dict[str, int]], int]:
    positions: Dict[str, Dict[str, int]] = {}
    ordered_panels: List[PanelSpec] = []
    grouped: Dict[str, PanelSpec] = {}

    for panel in section_panels:
        layer = _layer_from_panel_id(panel.panel_id)
        if layer in LAYER_ORDER and panel.panel_id.endswith("-latency"):
            grouped[layer] = panel

    for layer in LAYER_ORDER:
        panel = grouped.get(layer)
        if panel is not None:
            ordered_panels.append(panel)

    x = 0
    y = y_start
    row_height = 0
    for panel in ordered_panels:
        width = 6
        height = 5
        if x + width > 24:
            x = 0
            y += row_height + 1
            row_height = 0

        positions[panel.panel_id] = {"x": x, "y": y, "w": width, "h": height}
        x += width
        row_height = max(row_height, height)

    if ordered_panels:
        y += row_height

    return positions, y


def _layout_two_panel_page(
    page_panels: List[PanelSpec],
    y_start: int,
    first_panel_id: str,
    second_panel_id: str,
) -> Tuple[Dict[str, Dict[str, int]], int]:
    positions: Dict[str, Dict[str, int]] = {}
    by_id = {panel.panel_id: panel for panel in page_panels}

    first = by_id.get(first_panel_id)
    second = by_id.get(second_panel_id)

    if first is not None:
        positions[first.panel_id] = {"x": 0, "y": y_start, "w": 12, "h": 6}
    if second is not None:
        positions[second.panel_id] = {"x": 12, "y": y_start, "w": 12, "h": 6}

    return positions, y_start + 7


def _layout_change_rollout_page(
    page_panels: List[PanelSpec],
    y_start: int,
) -> Tuple[Dict[str, Dict[str, int]], int]:
    positions: Dict[str, Dict[str, int]] = {}
    by_id = {panel.panel_id: panel for panel in page_panels}

    panel_ids = [
        "change-rollout-health",
        "change-rollout-events",
        "change-rollout-capacity-diagnostic",
    ]

    x = 0
    for panel_id in panel_ids:
        panel = by_id.get(panel_id)
        if panel is None:
            continue
        positions[panel.panel_id] = {"x": x, "y": y_start, "w": 8, "h": 6}
        x += 8

    return positions, y_start + 7


def _build_grid_positions(panel_specs: List[PanelSpec]) -> Dict[str, Dict[str, int]]:
    positions: Dict[str, Dict[str, int]] = {}
    y = 0

    page_order: List[str] = []
    for panel in panel_specs:
        if panel.page_id not in page_order:
            page_order.append(panel.page_id)

    for page_id in page_order:
        page_panels = [panel for panel in panel_specs if panel.page_id == page_id]

        if page_id == "core-links":
            custom_positions, y = _layout_two_panel_page(
                page_panels,
                y,
                "core-links-health-request-success",
                "core-links-diagnostic-latency",
            )
            positions.update(custom_positions)
            y += 2
            continue

        if page_id == "dependencies":
            custom_positions, y = _layout_two_panel_page(
                page_panels,
                y,
                "dependencies-health",
                "dependencies-diagnostic-latency",
            )
            positions.update(custom_positions)
            y += 2
            continue

        if page_id == "error-analysis":
            custom_positions, y = _layout_two_panel_page(
                page_panels,
                y,
                "error-analysis-rate-volume",
                "error-analysis-diagnostic-topn",
            )
            positions.update(custom_positions)
            y += 2
            continue

        if page_id == "change-rollout-capacity":
            custom_positions, y = _layout_change_rollout_page(page_panels, y)
            positions.update(custom_positions)
            y += 2
            continue

        section_order: List[str] = []
        for panel in page_panels:
            if panel.section_id not in section_order:
                section_order.append(panel.section_id)

        for section_id in section_order:
            section_panels = [panel for panel in page_panels if panel.section_id == section_id]

            if section_id == "overview-row1-health-summary":
                custom_positions, y = _layout_overview_row1(section_panels, y)
                positions.update(custom_positions)
                continue

            if section_id == "overview-row2-health-sli-detail":
                custom_positions, y = _layout_overview_row2(section_panels, y)
                positions.update(custom_positions)
                continue

            if section_id == "overview-row3-event-overlay":
                custom_positions, y = _layout_overview_row3(section_panels, y)
                positions.update(custom_positions)
                continue

            if section_id == "service-layers-health":
                custom_positions, y = _layout_service_layers_health(section_panels, y)
                positions.update(custom_positions)
                continue

            if section_id == "service-layers-diagnostic":
                custom_positions, y = _layout_service_layers_diagnostic(section_panels, y)
                positions.update(custom_positions)
                continue

            x = 0
            row_height = 0
            for panel in section_panels:
                width, height = _panel_size(panel)
                if x + width > 24:
                    x = 0
                    y += row_height + 1
                    row_height = 0
                positions[panel.panel_id] = {"x": x, "y": y, "w": width, "h": height}
                x += width
                row_height = max(row_height, height)

            y += row_height + 1

        y += 2

    return positions


def build_grafana_dashboard(ia: DashboardIA, panel_specs: List[PanelSpec]) -> GrafanaDashboardDoc:
    positions = _build_grid_positions(panel_specs)
    panels: List[GrafanaPanelDTO] = []
    for spec in panel_specs:
        prefix = _zone_prefix(spec.zone_type)
        panels.append(
            GrafanaPanelDTO(
                panel_id=spec.panel_id,
                title=f"{prefix} {spec.title}",
                panel_type=spec.chart_type,
                expr=spec.metric_formula,
                grid_pos=positions[spec.panel_id],
                description=(
                    f"sli={spec.sli_link} path={spec.path_link} confidence={spec.confidence}"
                ),
                links=[{"title": spec.drilldown.title, "url": spec.drilldown.target}],
                threshold_steps=_threshold_steps(spec),
            )
        )

    tags = ["dashboard-design", "health", "diagnostic", "sli", "arkclaw-layout"]
    annotations = [
        {
            "name": "Alert Events",
            "datasource": {"type": "grafana", "uid": "-- Grafana --"},
            "enable": True,
            "iconColor": "red",
            "type": "dashboard",
        },
        {
            "name": "Change Events",
            "datasource": {"type": "grafana", "uid": "-- Grafana --"},
            "enable": True,
            "iconColor": "blue",
            "type": "dashboard",
        },
    ]

    return GrafanaDashboardDoc(
        title=ia.title,
        uid=_slug(ia.title)[:40],
        tags=tags,
        refresh="30s",
        time_from="now-6h",
        panels=panels,
        templating_variables=ia.global_variables,
        annotations=annotations,
    )

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from pathlib import Path
import json
from typing import Dict, List

from .models import DashboardIA, PanelSpec, ValidationReport


def _slug(text: str) -> str:
    chars = [c.lower() if c.isalnum() else "-" for c in text.strip()]
    slug = "".join(chars)
    while "--" in slug:
        slug = slug.replace("--", "-")
    return slug.strip("-") or "repo"


def write_outputs(
    out_base_dir: str,
    repo_slug: str,
    ia: DashboardIA,
    panel_specs: List[PanelSpec],
    grafana_dashboard: Dict[str, object],
    traceability: Dict[str, object],
    validation: ValidationReport,
    evidence_enriched: List[Dict[str, object]],
) -> str:
    outdir = Path(out_base_dir) / _slug(repo_slug)
    outdir.mkdir(parents=True, exist_ok=True)

    (outdir / "dashboard-ia.json").write_text(
        json.dumps(ia.to_dict(), ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (outdir / "panel-specs.json").write_text(
        json.dumps([panel.to_dict() for panel in panel_specs], ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (outdir / "grafana-dashboard.json").write_text(
        json.dumps(grafana_dashboard, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (outdir / "traceability.json").write_text(
        json.dumps(traceability, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (outdir / "validation-report.json").write_text(
        json.dumps(validation.to_dict(), ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (outdir / "evidence-index.enriched.json").write_text(
        json.dumps(evidence_enriched, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    return str(outdir)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import List

from .models import DashboardIA, IAPage, IASection, NormalizedInputs


def _overview_sections() -> List[IASection]:
    return [
        IASection(
            section_id="overview-row1-health-summary",
            name="宏观稳定性与处置效率",
            zone_type="health",
            purpose="展示整体可用率、SLO 达标率、错误预算与告警收敛效率",
        ),
        IASection(
            section_id="overview-row2-health-sli-detail",
            name="SLI/SLO 明细与趋势",
            zone_type="health",
            purpose="展示控制面与数据面 SLI 指标趋势和受损 TopN 维度",
        ),
        IASection(
            section_id="overview-row3-event-overlay",
            name="告警与变更事件",
            zone_type="event_overlay",
            purpose="展示告警、值班处置、可靠性事件与变更事件清单",
        ),
        IASection(
            section_id="overview-row4-diagnostic-component-health",
            name="核心组件诊断",
            zone_type="diagnostic",
            purpose="展示 API、模型、队列、Agent 等组件的诊断信号",
        ),
    ]


def _core_links_sections() -> List[IASection]:
    return [
        IASection(
            section_id="core-links-health",
            name="核心链路健康",
            zone_type="health",
            purpose="展示核心请求链路成功率与流量规模",
        ),
        IASection(
            section_id="core-links-diagnostic",
            name="链路诊断",
            zone_type="diagnostic",
            purpose="展示链路分段时延与错误热点",
        ),
    ]


def _service_layers_sections() -> List[IASection]:
    return [
        IASection(
            section_id="service-layers-health",
            name="分层健康总览",
            zone_type="health",
            purpose="按 Channel/Gateway、Session/Message、Queue、Agent、LLM、CronJob 分层观察稳定性",
        ),
        IASection(
            section_id="service-layers-diagnostic",
            name="分层诊断",
            zone_type="diagnostic",
            purpose="按层观察时延、错误和饱和度变化",
        ),
    ]


def _dependencies_sections() -> List[IASection]:
    return [
        IASection(
            section_id="dependencies-health",
            name="依赖健康",
            zone_type="health",
            purpose="展示强依赖可用率与依赖稳定性风险",
        ),
        IASection(
            section_id="dependencies-diagnostic",
            name="依赖诊断",
            zone_type="diagnostic",
            purpose="展示依赖调用时延、错误和资源容量瓶颈",
        ),
    ]


def _error_analysis_sections() -> List[IASection]:
    return [
        IASection(
            section_id="error-analysis-health",
            name="错误健康信号",
            zone_type="health",
            purpose="展示错误率、错误量与错误预算消耗",
        ),
        IASection(
            section_id="error-analysis-diagnostic",
            name="错误归因诊断",
            zone_type="diagnostic",
            purpose="按错误码、组件、租户、模型、渠道分解受损范围",
        ),
    ]


def _change_rollout_capacity_sections() -> List[IASection]:
    return [
        IASection(
            section_id="change-rollout-health",
            name="变更与灰度健康",
            zone_type="health",
            purpose="展示灰度窗口稳定性与回滚风险",
        ),
        IASection(
            section_id="change-rollout-events",
            name="告警/变更事件叠加",
            zone_type="event_overlay",
            purpose="在健康指标上叠加关键告警和变更事件",
        ),
        IASection(
            section_id="change-rollout-diagnostic",
            name="容量与资源诊断",
            zone_type="diagnostic",
            purpose="展示 CPU、内存、队列、连接池饱和度与值班收敛",
        ),
    ]


def build_dashboard_ia(inputs: NormalizedInputs, dashboard_title: str | None = None) -> DashboardIA:
    title = dashboard_title or f"{inputs.repo_slug} Dashboard Design"
    pages = [
        IAPage(
            page_id="overview",
            page_type="overview",
            name="总览页",
            purpose="展示服务整体稳定性水平、问题处置效率和质量",
            sections=_overview_sections(),
        ),
        IAPage(
            page_id="core-links",
            page_type="core_links",
            name="核心链路页",
            purpose="展示核心用户链路、控制面链路与数据面链路",
            sections=_core_links_sections(),
        ),
        IAPage(
            page_id="service-layers",
            page_type="service_layers",
            name="服务分层页",
            purpose="按服务层次进行健康和诊断分析",
            sections=_service_layers_sections(),
        ),
        IAPage(
            page_id="dependencies",
            page_type="dependencies",
            name="依赖资源页",
            purpose="展示依赖服务、数据库、缓存、队列等资源健康",
            sections=_dependencies_sections(),
        ),
        IAPage(
            page_id="error-analysis",
            page_type="error_analysis",
            name="错误分析页",
            purpose="定位错误来源与受损范围",
            sections=_error_analysis_sections(),
        ),
        IAPage(
            page_id="change-rollout-capacity",
            page_type="change_rollout_capacity",
            name="变更 / 灰度 / 容量页",
            purpose="关联变更、灰度、容量与告警事件",
            sections=_change_rollout_capacity_sections(),
        ),
    ]

    variables = ["service", "region", "claw", "tenant", "agent", "model", "channel", "cluster"]
    if inputs.focus_service:
        variables.append("focus_service")

    return DashboardIA(title=title, pages=pages, global_variables=variables)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from pathlib import Path
from typing import Any, Dict, Iterable, List

from .adapters import discover_code_metric_hints, load_arch_model, load_sli_model
from .models import (
    ArchPath,
    DependencyRisk,
    EvidenceItem,
    NormalizedInputs,
    ObservabilityGap,
    SLIIndicator,
)


def _as_list(value: Any) -> List[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def _pick_text(*values: Any, default: str = "") -> str:
    for value in values:
        if isinstance(value, str) and value.strip():
            return value.strip()
    return default


def _normalize_sli_indicator(item: Dict[str, Any], index: int) -> SLIIndicator:
    indicator_id = _pick_text(item.get("indicator_id"), item.get("id"), item.get("sli_name"), default=f"sli-{index}")
    name = _pick_text(
        item.get("name"),
        item.get("title"),
        item.get("indicator"),
        item.get("sli_name"),
        item.get("capability"),
        default=indicator_id,
    )
    sli_type = _pick_text(
        item.get("sli_type"),
        item.get("type"),
        item.get("category"),
        default="availability",
    )
    target = _pick_text(item.get("target"), item.get("objective"), item.get("slo"), item.get("target_slo"))
    formula = _pick_text(
        item.get("formula"),
        item.get("query"),
        item.get("metric_formula"),
        item.get("measurement"),
    )
    owner = _pick_text(item.get("owner"), item.get("service"), item.get("component"))
    return SLIIndicator(
        indicator_id=indicator_id,
        name=name,
        sli_type=sli_type,
        target=target,
        formula=formula,
        owner=owner,
    )


def _extract_sli_indicators(payload: Any) -> List[SLIIndicator]:
    candidates: List[Dict[str, Any]] = []

    if isinstance(payload, list):
        for item in payload:
            if isinstance(item, dict):
                candidates.append(item)
        return [_normalize_sli_indicator(item, index) for index, item in enumerate(candidates, start=1)]

    if not isinstance(payload, dict):
        return []

    for key in ["indicators", "sli_indicators", "slis", "specs"]:
        value = payload.get(key)
        if isinstance(value, list):
            for item in value:
                if isinstance(item, dict):
                    if isinstance(item.get("indicators"), list):
                        for nested in item["indicators"]:
                            if isinstance(nested, dict):
                                candidates.append(nested)
                    else:
                        candidates.append(item)

    items_value = payload.get("items")
    if not candidates and isinstance(items_value, list):
        for item in items_value:
            if isinstance(item, dict):
                candidates.append(item)

    if not candidates and payload:
        candidates.append(payload)

    return [_normalize_sli_indicator(item, index) for index, item in enumerate(candidates, start=1)]


def _normalize_path(item: Dict[str, Any], index: int, category: str) -> ArchPath:
    path_id = _pick_text(item.get("id"), item.get("path_id"), default=f"{category}-{index}")
    name = _pick_text(item.get("name"), item.get("title"), default=path_id)
    hops_raw = item.get("hops")
    hops: List[str]
    if isinstance(hops_raw, list):
        hops = [str(v) for v in hops_raw if str(v).strip()]
    elif isinstance(item.get("path"), list):
        hops = [str(v) for v in item["path"] if str(v).strip()]
    else:
        hops = []
    evidence = [str(v) for v in _as_list(item.get("evidence")) if str(v).strip()]
    return ArchPath(path_id=path_id, name=name, category=category, hops=hops, evidence=evidence)


def _extract_paths(topology_model: Dict[str, Any], key: str, category: str) -> List[ArchPath]:
    records = topology_model.get(key)
    if not isinstance(records, list):
        return []
    result: List[ArchPath] = []
    for index, item in enumerate(records, start=1):
        if isinstance(item, dict):
            result.append(_normalize_path(item, index, category))
    return result


def _extract_dependency_risks(topology_model: Dict[str, Any]) -> List[DependencyRisk]:
    risks = topology_model.get("failure_points")
    if not isinstance(risks, list):
        return []
    normalized: List[DependencyRisk] = []
    for item in risks:
        if not isinstance(item, dict):
            continue
        normalized.append(
            DependencyRisk(
                component=_pick_text(item.get("component"), item.get("service"), default="unknown"),
                risk=_pick_text(item.get("risk"), item.get("description"), default="unknown risk"),
                impact=_pick_text(item.get("impact"), item.get("severity"), default="unknown impact"),
                evidence=[str(v) for v in _as_list(item.get("evidence")) if str(v).strip()],
            )
        )
    return normalized


def _extract_observability_gaps(topology_model: Dict[str, Any]) -> List[ObservabilityGap]:
    gaps = topology_model.get("observability_hook_points")
    if not isinstance(gaps, list):
        return []
    normalized: List[ObservabilityGap] = []
    for item in gaps:
        if not isinstance(item, dict):
            continue
        normalized.append(
            ObservabilityGap(
                component=_pick_text(item.get("component"), default="unknown"),
                gap_type=_pick_text(item.get("gap_type"), item.get("hook_type"), default="metric"),
                missing_signal=_pick_text(item.get("missing_signal"), item.get("description"), default="unknown"),
                suggestion=_pick_text(item.get("suggestion"), item.get("target"), default="add observability"),
                evidence=[str(v) for v in _as_list(item.get("evidence")) if str(v).strip()],
            )
        )
    return normalized


def _extract_evidence_items(evidence_payload: Any) -> List[EvidenceItem]:
    items = evidence_payload if isinstance(evidence_payload, list) else []
    normalized: List[EvidenceItem] = []
    for index, item in enumerate(items, start=1):
        if not isinstance(item, dict):
            continue
        source_path = _pick_text(item.get("source"), item.get("source_path"), default="unknown")
        normalized.append(
            EvidenceItem(
                evidence_id=_pick_text(item.get("id"), default=f"ev-{index}"),
                source_type=_pick_text(item.get("kind"), item.get("source_type"), default="unknown"),
                source_path=source_path,
                locator=_pick_text(item.get("locator"), default=source_path),
                summary=_pick_text(item.get("summary"), default=source_path),
            )
        )
    return normalized


def _derive_repo_slug(repo: str) -> str:
    return Path(repo).name.strip() or "repo"


def normalize_inputs(
    sli_model_path: str,
    arch_model_dir: str,
    repo: str,
    focus_service: str | None = None,
    offline: bool = False,
) -> NormalizedInputs:
    _ = offline

    sli_payload = load_sli_model(sli_model_path)
    arch_payload = load_arch_model(arch_model_dir)

    topology_model = arch_payload["topology_model"]
    sli_indicators = _extract_sli_indicators(sli_payload)
    request_paths = _extract_paths(topology_model, "request_paths", "request")
    async_paths = _extract_paths(topology_model, "async_paths", "async")
    dependency_risks = _extract_dependency_risks(topology_model)
    observability_gaps = _extract_observability_gaps(topology_model)
    evidence_items = _extract_evidence_items(arch_payload.get("evidence_index", []))

    code_metric_hints = discover_code_metric_hints(repo)

    return NormalizedInputs(
        repo_slug=_derive_repo_slug(repo),
        sli_indicators=sli_indicators,
        request_paths=request_paths,
        async_paths=async_paths,
        dependency_risks=dependency_risks,
        observability_gaps=observability_gaps,
        code_metric_hints=code_metric_hints,
        evidence_items=evidence_items,
        focus_service=focus_service,
    )

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional


@dataclass
class SLIIndicator:
    indicator_id: str
    name: str
    sli_type: str
    target: str = ""
    formula: str = ""
    owner: str = ""


@dataclass
class ArchPath:
    path_id: str
    name: str
    category: str
    hops: List[str] = field(default_factory=list)
    evidence: List[str] = field(default_factory=list)


@dataclass
class DependencyRisk:
    component: str
    risk: str
    impact: str
    evidence: List[str] = field(default_factory=list)


@dataclass
class ObservabilityGap:
    component: str
    gap_type: str
    missing_signal: str
    suggestion: str
    evidence: List[str] = field(default_factory=list)


@dataclass
class EvidenceItem:
    evidence_id: str
    source_type: str
    source_path: str
    locator: str
    summary: str


@dataclass
class TraceLink:
    target_type: str
    target_id: str
    evidence_ids: List[str] = field(default_factory=list)
    rationale: str = ""


@dataclass
class NormalizedInputs:
    repo_slug: str
    sli_indicators: List[SLIIndicator] = field(default_factory=list)
    request_paths: List[ArchPath] = field(default_factory=list)
    async_paths: List[ArchPath] = field(default_factory=list)
    dependency_risks: List[DependencyRisk] = field(default_factory=list)
    observability_gaps: List[ObservabilityGap] = field(default_factory=list)
    code_metric_hints: List[str] = field(default_factory=list)
    evidence_items: List[EvidenceItem] = field(default_factory=list)
    focus_service: Optional[str] = None


@dataclass
class IASection:
    section_id: str
    name: str
    zone_type: str
    purpose: str


@dataclass
class IAPage:
    page_id: str
    page_type: str
    name: str
    purpose: str
    sections: List[IASection] = field(default_factory=list)


@dataclass
class DashboardIA:
    title: str
    pages: List[IAPage] = field(default_factory=list)
    global_variables: List[str] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class ThresholdRule:
    color: str
    operator: str
    value: float


@dataclass
class DrillDownLink:
    title: str
    target: str


@dataclass
class PanelSpec:
    panel_id: str
    page_id: str
    section_id: str
    title: str
    chart_type: str
    metric_formula: str
    dimensions: List[str]
    refresh_interval: str
    thresholds: List[ThresholdRule]
    drilldown: DrillDownLink
    sli_link: str
    path_link: str
    zone_type: str
    confidence: str
    evidence_refs: List[str] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        data = asdict(self)
        data["thresholds"] = [asdict(item) for item in self.thresholds]
        data["drilldown"] = asdict(self.drilldown)
        return data


@dataclass
class GrafanaPanelDTO:
    panel_id: str
    title: str
    panel_type: str
    expr: str
    grid_pos: Dict[str, int]
    description: str
    links: List[Dict[str, str]] = field(default_factory=list)
    threshold_steps: List[Dict[str, Any]] = field(default_factory=list)


@dataclass
class GrafanaDashboardDoc:
    title: str
    uid: str
    tags: List[str] = field(default_factory=list)
    refresh: str = "30s"
    time_from: str = "now-6h"
    panels: List[GrafanaPanelDTO] = field(default_factory=list)
    templating_variables: List[str] = field(default_factory=list)
    annotations: List[Dict[str, Any]] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "title": self.title,
            "uid": self.uid,
            "tags": self.tags,
            "refresh": self.refresh,
            "time": {"from": self.time_from, "to": "now"},
            "templating": {
                "list": [
                    {
                        "name": name,
                        "label": name,
                        "type": "textbox",
                        "query": "",
                        "current": {"text": "all", "value": "all"},
                    }
                    for name in self.templating_variables
                ]
            },
            "annotations": {"list": self.annotations},
            "panels": [
                {
                    "id": index + 1,
                    "title": panel.title,
                    "type": panel.panel_type,
                    "gridPos": panel.grid_pos,
                    "targets": [{"refId": "A", "expr": panel.expr}],
                    "fieldConfig": {
                        "defaults": {
                            "thresholds": {
                                "mode": "absolute",
                                "steps": panel.threshold_steps,
                            }
                        },
                        "overrides": [],
                    },
                    "description": panel.description,
                    "links": panel.links,
                }
                for index, panel in enumerate(self.panels)
            ],
            "schemaVersion": 39,
            "version": 1,
        }


@dataclass
class ValidationIssue:
    level: str
    rule: str
    message: str


@dataclass
class ValidationReport:
    passed: bool
    summary: Dict[str, int]
    issues: List[ValidationIssue] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "passed": self.passed,
            "summary": self.summary,
            "issues": [asdict(issue) for issue in self.issues],
        }


@dataclass
class PipelineResult:
    output_dir: str
    panel_count: int
    placeholder_panel_count: int
    page_count: int
    validation_passed: bool

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Iterable, List, Tuple

from .models import (
    DashboardIA,
    DrillDownLink,
    NormalizedInputs,
    PanelSpec,
    ThresholdRule,
)


def _select_metric_formula(
    keyword: str,
    code_metric_hints: List[str],
    fallback: str,
) -> Tuple[str, str]:
    for name in code_metric_hints:
        if keyword in name.lower():
            return f"sum(rate({name}[5m]))", "high"
    return fallback, "placeholder"


def _default_thresholds(zone_type: str) -> List[ThresholdRule]:
    if zone_type == "health":
        return [
            ThresholdRule(color="green", operator=">=", value=99),
            ThresholdRule(color="yellow", operator="<", value=99),
            ThresholdRule(color="red", operator="<", value=95),
        ]
    if zone_type == "event_overlay":
        return [
            ThresholdRule(color="green", operator="<=", value=5),
            ThresholdRule(color="yellow", operator=">", value=5),
            ThresholdRule(color="red", operator=">", value=20),
        ]
    return [
        ThresholdRule(color="green", operator="<=", value=1),
        ThresholdRule(color="yellow", operator=">", value=1),
        ThresholdRule(color="red", operator=">", value=5),
    ]


def _panel(
    panel_id: str,
    page_id: str,
    section_id: str,
    zone_type: str,
    title: str,
    chart_type: str,
    metric_formula: str,
    confidence: str,
    sli_link: str,
    path_link: str,
    evidence_refs: Iterable[str],
    refresh_interval: str = "30s",
    dimensions: List[str] | None = None,
) -> PanelSpec:
    return PanelSpec(
        panel_id=panel_id,
        page_id=page_id,
        section_id=section_id,
        title=title,
        chart_type=chart_type,
        metric_formula=metric_formula,
        dimensions=dimensions or ["service", "region", "tenant"],
        refresh_interval=refresh_interval,
        thresholds=_default_thresholds(zone_type),
        drilldown=DrillDownLink(title="View details", target=f"page:{page_id}"),
        sli_link=sli_link,
        path_link=path_link,
        zone_type=zone_type,
        confidence=confidence,
        evidence_refs=list(evidence_refs),
    )


def compile_panel_specs(inputs: NormalizedInputs, ia: DashboardIA) -> List[PanelSpec]:
    panels: List[PanelSpec] = []

    evidence_ids = [item.evidence_id for item in inputs.evidence_items[:8]]
    main_path = inputs.request_paths[0].path_id if inputs.request_paths else "request-main"
    main_sli = inputs.sli_indicators[0].indicator_id if inputs.sli_indicators else "sli-availability"

    availability_formula, availability_conf = _select_metric_formula(
        "success_rate",
        inputs.code_metric_hints,
        fallback="(sum(rate(service_success_total[5m])) / sum(rate(service_requests_total[5m]))) * 100",
    )
    slo_formula, slo_conf = _select_metric_formula(
        "slo",
        inputs.code_metric_hints,
        fallback="(sum(rate(slo_good_total[5m])) / sum(rate(slo_total[5m]))) * 100",
    )
    error_budget_formula, error_budget_conf = _select_metric_formula(
        "error",
        inputs.code_metric_hints,
        fallback="(sum(rate(service_error_total[5m])) / sum(rate(service_requests_total[5m]))) * 100",
    )

    panels.extend(
        [
            _panel(
                panel_id="overview-kpi-availability",
                page_id="overview",
                section_id="overview-row1-health-summary",
                zone_type="health",
                title="整体可用率",
                chart_type="stat",
                metric_formula=availability_formula,
                confidence=availability_conf,
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["service"],
            ),
            _panel(
                panel_id="overview-kpi-slo-attainment",
                page_id="overview",
                section_id="overview-row1-health-summary",
                zone_type="health",
                title="SLO 达标率",
                chart_type="stat",
                metric_formula=slo_formula,
                confidence=slo_conf,
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["service"],
            ),
            _panel(
                panel_id="overview-kpi-error-budget",
                page_id="overview",
                section_id="overview-row1-health-summary",
                zone_type="health",
                title="错误预算消耗",
                chart_type="stat",
                metric_formula=error_budget_formula,
                confidence=error_budget_conf,
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["service"],
            ),
            _panel(
                panel_id="overview-kpi-active-alerts",
                page_id="overview",
                section_id="overview-row1-health-summary",
                zone_type="health",
                title="今日告警数",
                chart_type="stat",
                metric_formula="sum(increase(alert_events_total[1d]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["severity"],
            ),
            _panel(
                panel_id="overview-kpi-oncall-topn",
                page_id="overview",
                section_id="overview-row1-health-summary",
                zone_type="diagnostic",
                title="TOPN Oncall / 工单收敛率",
                chart_type="table",
                metric_formula="topk(10, sum by(oncall)(increase(oncall_closed_incidents_total[1d])))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["oncall"],
            ),
        ]
    )

    panels.extend(
        [
            _panel(
                panel_id="overview-sli-scorecards",
                page_id="overview",
                section_id="overview-row2-health-sli-detail",
                zone_type="health",
                title="控制面/数据面 SLI 评分卡",
                chart_type="table",
                metric_formula="sum by(sli)(rate(sli_good_total[5m])) / sum by(sli)(rate(sli_total[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["sli", "plane"],
            ),
            _panel(
                panel_id="overview-sli-trend",
                page_id="overview",
                section_id="overview-row2-health-sli-detail",
                zone_type="health",
                title="可用率趋势图",
                chart_type="timeseries",
                metric_formula="sum(rate(control_plane_requests_total[5m])) / sum(rate(control_plane_requests_total[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["service", "plane"],
            ),
            _panel(
                panel_id="overview-topn-impact-tenant",
                page_id="overview",
                section_id="overview-row2-health-sli-detail",
                zone_type="diagnostic",
                title="TOPN 受损租户（成功率下降）",
                chart_type="table",
                metric_formula="topk(10, (1 - (sum by(tenant)(rate(request_success_total[5m])) / sum by(tenant)(rate(request_total[5m])))) * 100)",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["tenant"],
            ),
            _panel(
                panel_id="overview-topn-impact-model",
                page_id="overview",
                section_id="overview-row2-health-sli-detail",
                zone_type="diagnostic",
                title="TOPN 受损模型（成功率下降）",
                chart_type="table",
                metric_formula="topk(10, (1 - (sum by(model)(rate(request_success_total[5m])) / sum by(model)(rate(request_total[5m])))) * 100)",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["model"],
            ),
            _panel(
                panel_id="overview-topn-impact-channel",
                page_id="overview",
                section_id="overview-row2-health-sli-detail",
                zone_type="diagnostic",
                title="TOPN 受损渠道（成功率下降）",
                chart_type="table",
                metric_formula="topk(10, (1 - (sum by(channel)(rate(request_success_total[5m])) / sum by(channel)(rate(request_total[5m])))) * 100)",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["channel"],
            ),
            _panel(
                panel_id="overview-topn-impact-agent",
                page_id="overview",
                section_id="overview-row2-health-sli-detail",
                zone_type="diagnostic",
                title="TOPN 受损 Agent（成功率下降）",
                chart_type="table",
                metric_formula="topk(10, (1 - (sum by(agent)(rate(request_success_total[5m])) / sum by(agent)(rate(request_total[5m])))) * 100)",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["agent"],
            ),
        ]
    )

    panels.extend(
        [
            _panel(
                panel_id="overview-events-alert-stream",
                page_id="overview",
                section_id="overview-row3-event-overlay",
                zone_type="event_overlay",
                title="告警事件流",
                chart_type="table",
                metric_formula="topk(50, increase(alert_events_total[1h]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["severity", "service"],
            ),
            _panel(
                panel_id="overview-events-reliability",
                page_id="overview",
                section_id="overview-row3-event-overlay",
                zone_type="event_overlay",
                title="稳定性事件（故障/恢复）",
                chart_type="table",
                metric_formula="topk(50, increase(reliability_events_total[1h]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["event_type", "service"],
            ),
            _panel(
                panel_id="overview-events-change",
                page_id="overview",
                section_id="overview-row3-event-overlay",
                zone_type="event_overlay",
                title="变更事件",
                chart_type="table",
                metric_formula="topk(50, increase(change_events_total[1h]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["change_type", "service"],
            ),
        ]
    )

    layer_panels = [
        ("api", "API（网络层）", "api_success_total", "api_latency_ms"),
        ("gateway", "Channel / Gateway", "gateway_success_total", "gateway_latency_ms"),
        ("conversation", "Conversation / Message", "conversation_success_total", "conversation_latency_ms"),
        ("queue", "Message / Queue", "queue_success_total", "queue_latency_ms"),
        ("agent", "Agent", "agent_success_total", "agent_latency_ms"),
        ("llm", "LLM / Tool / Context", "llm_success_total", "llm_latency_ms"),
        ("cron", "CronJob / Skills", "cron_success_total", "cron_latency_ms"),
    ]

    for layer_id, layer_name, success_metric, latency_metric in layer_panels:
        panels.append(
            _panel(
                panel_id=f"service-layers-{layer_id}-scorecards",
                page_id="service-layers",
                section_id="service-layers-health",
                zone_type="health",
                title=f"{layer_name} 指标卡",
                chart_type="table",
                metric_formula=f"sum by(metric)(rate({success_metric}[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["metric", "region"],
            )
        )
        panels.append(
            _panel(
                panel_id=f"service-layers-{layer_id}-availability",
                page_id="service-layers",
                section_id="service-layers-health",
                zone_type="health",
                title=f"{layer_name} 可用率趋势",
                chart_type="timeseries",
                metric_formula=f"sum(rate({success_metric}[5m])) / sum(rate({success_metric.replace('_success_total', '_requests_total')}[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["service", "region"],
            )
        )
        panels.append(
            _panel(
                panel_id=f"service-layers-{layer_id}-latency",
                page_id="service-layers",
                section_id="service-layers-diagnostic",
                zone_type="diagnostic",
                title=f"{layer_name} 延迟 P95/P99",
                chart_type="timeseries",
                metric_formula=f"histogram_quantile(0.95, sum(rate({latency_metric}_bucket[5m])) by (le, service))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["service", "region"],
            )
        )

    panels.extend(
        [
            _panel(
                panel_id="core-links-health-request-success",
                page_id="core-links",
                section_id="core-links-health",
                zone_type="health",
                title="核心链路成功率",
                chart_type="timeseries",
                metric_formula="sum(rate(request_success_total[5m])) / sum(rate(request_total[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["path", "service"],
            ),
            _panel(
                panel_id="core-links-diagnostic-latency",
                page_id="core-links",
                section_id="core-links-diagnostic",
                zone_type="diagnostic",
                title="链路分段时延 P99",
                chart_type="timeseries",
                metric_formula="histogram_quantile(0.99, sum(rate(request_duration_ms_bucket[5m])) by (le, hop))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["hop"],
            ),
            _panel(
                panel_id="dependencies-health",
                page_id="dependencies",
                section_id="dependencies-health",
                zone_type="health",
                title="强依赖可用率",
                chart_type="table",
                metric_formula="sum by(dependency)(rate(dependency_success_total[5m])) / sum by(dependency)(rate(dependency_total[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["dependency"],
            ),
            _panel(
                panel_id="dependencies-diagnostic-latency",
                page_id="dependencies",
                section_id="dependencies-diagnostic",
                zone_type="diagnostic",
                title="依赖延迟与容量热点",
                chart_type="timeseries",
                metric_formula="topk(10, histogram_quantile(0.99, sum(rate(dependency_duration_ms_bucket[5m])) by (le, dependency)))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["dependency"],
            ),
            _panel(
                panel_id="error-analysis-rate-volume",
                page_id="error-analysis",
                section_id="error-analysis-health",
                zone_type="health",
                title="错误率与错误量",
                chart_type="timeseries",
                metric_formula="sum(rate(error_total[5m])) / sum(rate(request_total[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
            ),
            _panel(
                panel_id="error-analysis-diagnostic-topn",
                page_id="error-analysis",
                section_id="error-analysis-diagnostic",
                zone_type="diagnostic",
                title="错误归因 TopN",
                chart_type="table",
                metric_formula="topk(20, sum by(error_code, component)(increase(error_total[1h])))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["error_code", "component"],
            ),
            _panel(
                panel_id="change-rollout-health",
                page_id="change-rollout-capacity",
                section_id="change-rollout-health",
                zone_type="health",
                title="灰度窗口稳定性",
                chart_type="timeseries",
                metric_formula="sum(rate(canary_success_total[5m])) / sum(rate(canary_total[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
            ),
            _panel(
                panel_id="change-rollout-events",
                page_id="change-rollout-capacity",
                section_id="change-rollout-events",
                zone_type="event_overlay",
                title="变更/告警事件叠加",
                chart_type="state-timeline",
                metric_formula="sum(rate(alert_events_total[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["event_type", "source"],
            ),
            _panel(
                panel_id="change-rollout-capacity-diagnostic",
                page_id="change-rollout-capacity",
                section_id="change-rollout-diagnostic",
                zone_type="diagnostic",
                title="容量与饱和度诊断",
                chart_type="timeseries",
                metric_formula="sum(rate(queue_backlog_count[5m]))",
                confidence="placeholder",
                sli_link=main_sli,
                path_link=main_path,
                evidence_refs=evidence_ids,
                dimensions=["resource", "cluster"],
            ),
        ]
    )

    by_page = {page.page_id: page for page in ia.pages}
    valid_panels = [panel for panel in panels if panel.page_id in by_page]

    for panel in valid_panels:
        if panel.sli_link == "sli-availability" and inputs.sli_indicators:
            panel.sli_link = inputs.sli_indicators[0].indicator_id
        if panel.path_link == "request-main" and inputs.request_paths:
            panel.path_link = inputs.request_paths[0].path_id

    return valid_panels