Byted Sol Stability Grafana Dashboard Assembly

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-assembly"
version = "0.1.0"
description = "Assemble and integration-validate Grafana dashboards from SLI and metric mapping inputs"
readme = "SKILL.md"
requires-python = ">=3.10"
dependencies = ["pyyaml>=6.0"]

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

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

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

from .pipeline import PipelineOptions, run_pipeline

__all__ = ["PipelineOptions", "run_pipeline"]

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

from __future__ import annotations

from typing import List

from .models import AcceptanceReport, PanelFinding, TargetValidation


def build_acceptance_report(validations: List[TargetValidation], findings: List[PanelFinding]) -> AcceptanceReport:
    total = len(validations)
    success = sum(1 for item in validations if item.status == "success")
    success_rate = round((success / total), 3) if total else 0.0

    empty_panels = {item.panel_id for item in findings if item.finding_type == "no_data"}
    error_query_count = sum(1 for item in validations if item.status == "error")

    manual_items = []
    for finding in findings:
        if finding.severity in {"warning", "blocker"}:
            manual_items.append(f"panel {finding.panel_id} {finding.finding_type}: {finding.message}")

    blocker_count = sum(1 for finding in findings if finding.severity == "blocker")
    overall_pass = (
        success_rate >= 0.95
        and error_query_count == 0
        and len(empty_panels) <= 1
        and blocker_count == 0
    )

    return AcceptanceReport(
        success_rate=success_rate,
        empty_panels_count=len(empty_panels),
        error_query_count=error_query_count,
        manual_confirmation_items=manual_items,
        overall_pass=overall_pass,
    )

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

from .catalog_adapter import load_metrics_catalog
from .dashboard_adapter import load_existing_dashboard
from .log_adapter import load_log_dict
from .mapping_adapter import load_metric_mapping_spec
from .sli_adapter import load_sli_spec
from .trace_adapter import load_trace_spans

__all__ = [
    "load_sli_spec",
    "load_metric_mapping_spec",
    "load_metrics_catalog",
    "load_log_dict",
    "load_trace_spans",
    "load_existing_dashboard",
]

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

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


def load_metrics_catalog(path: str) -> Any:
    target = Path(path)
    if not target.exists() or not target.is_file():
        raise ValueError(f"metrics catalog path not found: {target}")
    return json.loads(target.read_text(encoding="utf-8"))

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

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


def load_existing_dashboard(path: str) -> Any:
    target = Path(path)
    if not target.exists() or not target.is_file():
        raise ValueError(f"existing dashboard path not found: {target}")
    return json.loads(target.read_text(encoding="utf-8"))

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

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


def load_log_dict(path: str) -> Any:
    target = Path(path)
    if not target.exists() or not target.is_file():
        raise ValueError(f"log dict path not found: {target}")
    return json.loads(target.read_text(encoding="utf-8"))

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

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


def load_metric_mapping_spec(path: str) -> Any:
    target = Path(path)
    if not target.exists() or not target.is_file():
        raise ValueError(f"metric mapping spec path not found: {target}")
    return json.loads(target.read_text(encoding="utf-8"))

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

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


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


def load_sli_spec(path: str) -> Any:
    target = Path(path)
    if not target.exists():
        raise ValueError(f"sli spec path not found: {target}")
    if target.is_dir():
        target = _pick_json_file(target)
    return json.loads(target.read_text(encoding="utf-8"))

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

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


def load_trace_spans(path: str) -> Any:
    target = Path(path)
    if not target.exists() or not target.is_file():
        raise ValueError(f"trace spans path not found: {target}")
    return json.loads(target.read_text(encoding="utf-8"))

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

from __future__ import annotations

import re
from copy import deepcopy
from typing import Dict, List, Optional, Tuple

from .models import FixAction, MetricMappingItem, NormalizedAssemblyInputs, PanelFinding, TargetValidation


METRIC_TOKEN = re.compile(r"\b([a-zA-Z_:][a-zA-Z0-9_:]*)\b")


def _tokens(text: str) -> set[str]:
    return {token for token in re.findall(r"[a-z0-9_]+", text.lower()) if len(token) >= 3}


def _best_mapping(panel_title: str, query: str, mapping_items: List[MetricMappingItem]) -> Optional[MetricMappingItem]:
    title_tokens = _tokens(panel_title)
    query_tokens = _tokens(query)

    scored: List[Tuple[MetricMappingItem, float]] = []
    for item in mapping_items:
        sli_tokens = _tokens(item.sli_name)
        score = len(title_tokens & sli_tokens) * 0.3 + len(query_tokens & sli_tokens) * 0.2
        if item.chosen_metric and item.chosen_metric in query:
            score += 0.7
        scored.append((item, score))

    scored.sort(key=lambda pair: pair[1], reverse=True)
    if not scored or scored[0][1] <= 0:
        return None
    return scored[0][0]


def _replace_first_metric(query: str, metric_name: str) -> str:
    reserved = {
        "sum",
        "avg",
        "min",
        "max",
        "count",
        "rate",
        "irate",
        "increase",
        "histogram_quantile",
        "topk",
        "bottomk",
        "clamp_min",
        "clamp_max",
        "by",
        "without",
    }

    def repl(match: re.Match[str]) -> str:
        token = match.group(1)
        if token in reserved:
            return token
        repl.called = True
        return metric_name

    repl.called = False  # type: ignore[attr-defined]
    replaced = METRIC_TOKEN.sub(repl, query, count=1)
    return replaced if repl.called else query


def _add_missing_labels(query: str, labels: List[str]) -> str:
    if not labels:
        return query
    if "by (" in query:
        return query
    grouped = ",".join(labels)
    if any(token in query for token in ["sum(", "avg(", "min(", "max(", "count("]):
        return f"sum by ({grouped}) ({query})"
    return query


def _ensure_default_service_variable(dashboard: Dict[str, object]) -> bool:
    templating = dashboard.setdefault("templating", {})
    if not isinstance(templating, dict):
        return False
    listing = templating.setdefault("list", [])
    if not isinstance(listing, list):
        return False

    if any(isinstance(item, dict) and item.get("name") == "service" for item in listing):
        return False

    listing.append(
        {
            "name": "service",
            "type": "query",
            "label": "service",
            "query": "label_values(up,service)",
            "refresh": 1,
            "hide": 0,
            "includeAll": False,
            "multi": False,
        }
    )
    return True


def apply_auto_repair(
    inputs: NormalizedAssemblyInputs,
    dashboard: Dict[str, object],
    validations: List[TargetValidation],
    findings: List[PanelFinding],
) -> Tuple[Dict[str, object], List[FixAction]]:
    patched = deepcopy(dashboard)
    actions: List[FixAction] = []

    if _ensure_default_service_variable(patched):
        actions.append(
            FixAction(
                panel_id=-1,
                panel_title="dashboard",
                target_index=-1,
                action="add_default_variable",
                before="",
                after="$service",
                reason="add missing default service variable",
                confidence=0.95,
            )
        )

    finding_by_target = {(item.panel_id, item.target_index): item for item in findings if item.target_index is not None}

    for panel in patched.get("panels", []):
        if not isinstance(panel, dict):
            continue
        panel_id = int(panel.get("id") or -1)
        panel_title = str(panel.get("title") or "Untitled")
        targets = panel.get("targets") if isinstance(panel.get("targets"), list) else []

        for index, target in enumerate(targets):
            if not isinstance(target, dict):
                continue
            validation = next((item for item in validations if item.panel_id == panel_id and item.target_index == index), None)
            if validation is None or validation.status == "success":
                continue

            before = str(target.get("expr") or target.get("query") or "")
            mapping = _best_mapping(panel_title, before, inputs.mapping_items)
            candidate = before

            if mapping and mapping.query_template and (not validation.executable or not validation.aggregation_ok):
                candidate = mapping.query_template
                if candidate != before:
                    actions.append(
                        FixAction(
                            panel_id=panel_id,
                            panel_title=panel_title,
                            target_index=index,
                            action="rewrite_query",
                            before=before,
                            after=candidate,
                            reason="repair query execution or aggregation semantics",
                            confidence=0.85,
                        )
                    )
            elif mapping and not validation.labels_ok:
                candidate = _add_missing_labels(before, mapping.dimensions)
                if candidate != before:
                    actions.append(
                        FixAction(
                            panel_id=panel_id,
                            panel_title=panel_title,
                            target_index=index,
                            action="adjust_labels",
                            before=before,
                            after=candidate,
                            reason="inject required grouping labels",
                            confidence=0.73,
                        )
                    )
            elif mapping and mapping.chosen_metric and mapping.chosen_metric not in before:
                candidate = _replace_first_metric(before, mapping.chosen_metric)
                if candidate != before:
                    actions.append(
                        FixAction(
                            panel_id=panel_id,
                            panel_title=panel_title,
                            target_index=index,
                            action="replace_metric_name",
                            before=before,
                            after=candidate,
                            reason="replace with mapped chosen metric",
                            confidence=0.78,
                        )
                    )

            if candidate == before and validation.status == "error":
                panel_type_before = str(panel.get("type") or "timeseries")
                if panel_type_before not in {"timeseries", "table"}:
                    panel["type"] = "timeseries"
                    actions.append(
                        FixAction(
                            panel_id=panel_id,
                            panel_title=panel_title,
                            target_index=index,
                            action="downgrade_panel_type",
                            before=panel_type_before,
                            after="timeseries",
                            reason="fallback panel type for unstable query result",
                            confidence=0.6,
                        )
                    )

            if "expr" in target:
                target["expr"] = candidate
            else:
                target["query"] = candidate

            finding = finding_by_target.get((panel_id, index))
            if finding and finding.finding_type == "no_data" and panel.get("description"):
                panel["description"] = str(panel.get("description")).strip()

    return patched, actions

# 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="Assemble and integration-validate Grafana dashboard JSON")
    parser.add_argument("--sli-spec", required=True, help="Path to SLI spec JSON")
    parser.add_argument("--metric-mapping-spec", required=True, help="Path to metric mapping spec JSON")
    parser.add_argument("--metrics-catalog", required=True, help="Path to metrics catalog JSON")
    parser.add_argument("--log-dict", required=True, help="Path to log field dictionary JSON")
    parser.add_argument("--trace-spans", required=True, help="Path to trace span names JSON")
    parser.add_argument("--existing-dashboard", required=True, help="Path to existing dashboard JSON")
    parser.add_argument("--grafana-url", default="", help="Grafana base URL")
    parser.add_argument("--grafana-token", default="", help="Grafana token")
    parser.add_argument("--datasource-uid", default="", help="Grafana datasource uid")
    parser.add_argument("--prom-url", default="", help="Prometheus base URL")
    parser.add_argument("--prom-bearer", default="", help="Prometheus bearer token")
    parser.add_argument("--prom-username", default="", help="Prometheus basic auth username")
    parser.add_argument("--prom-password", default="", help="Prometheus basic auth password")
    parser.add_argument("--time-range", default="now-6h,now", help="Time range like now-6h,now")
    parser.add_argument("--out-dir", default="output", help="Output base directory")
    parser.add_argument("--focus-service", help="Optional focus service")
    parser.add_argument("--offline", action="store_true", help="Use local checks only")
    parser.add_argument("--max-repair-rounds", type=int, default=2, help="Auto-repair rounds")
    return parser


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

    required_paths = [
        Path(args.sli_spec),
        Path(args.metric_mapping_spec),
        Path(args.metrics_catalog),
        Path(args.log_dict),
        Path(args.trace_spans),
        Path(args.existing_dashboard),
    ]
    for path in required_paths:
        if not path.exists():
            raise SystemExit(f"required path not found: {path}")

    result = run_pipeline(
        sli_spec=args.sli_spec,
        metric_mapping_spec=args.metric_mapping_spec,
        metrics_catalog=args.metrics_catalog,
        log_dict=args.log_dict,
        trace_spans=args.trace_spans,
        existing_dashboard=args.existing_dashboard,
        options=PipelineOptions(
            out_dir=args.out_dir,
            focus_service=args.focus_service,
            offline=args.offline,
            grafana_url=args.grafana_url,
            grafana_token=args.grafana_token,
            datasource_uid=args.datasource_uid,
            prom_url=args.prom_url,
            prom_bearer=args.prom_bearer,
            prom_username=args.prom_username,
            prom_password=args.prom_password,
            time_range=args.time_range,
            max_repair_rounds=max(1, args.max_repair_rounds),
        ),
    )

    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

import base64
import json
import urllib.error
import urllib.parse
import urllib.request
from typing import Dict, Iterable, List

from .models import ConnectivityCheck, RuntimeConfig


KNOWN_DATASOURCES = {"prometheus", "loki", "tempo", "internal_tsdb", "mixed", "grafana"}


def _auth_headers(runtime: RuntimeConfig) -> Dict[str, str]:
    headers: Dict[str, str] = {}
    if runtime.prom_bearer:
        headers["Authorization"] = f"Bearer {runtime.prom_bearer}"
    elif runtime.prom_username and runtime.prom_password:
        token = base64.b64encode(f"{runtime.prom_username}:{runtime.prom_password}".encode("utf-8")).decode("utf-8")
        headers["Authorization"] = f"Basic {token}"
    return headers


def _check_prometheus_query_api(runtime: RuntimeConfig) -> tuple[bool, str]:
    if not runtime.prom_url:
        return False, "prom_url missing"

    base = runtime.prom_url.rstrip("/")
    url = f"{base}/api/v1/query?{urllib.parse.urlencode({'query': '1'})}"
    request = urllib.request.Request(url, method="GET", headers=_auth_headers(runtime))
    try:
        with urllib.request.urlopen(request, timeout=8) as response:
            payload = json.loads(response.read().decode("utf-8"))
            if payload.get("status") in {None, "success"}:
                return True, "prometheus query api reachable"
            return False, f"prometheus status={payload.get('status')}"
    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
        return False, f"prometheus query api unreachable: {exc}"


def _check_grafana_query_api(runtime: RuntimeConfig) -> tuple[bool, str]:
    if not (runtime.grafana_url and runtime.grafana_token and runtime.datasource_uid):
        return False, "grafana credentials or datasource uid missing"

    url = runtime.grafana_url.rstrip("/") + "/api/ds/query"
    payload = {
        "queries": [
            {
                "refId": "A",
                "datasource": {"uid": runtime.datasource_uid},
                "datasourceUid": runtime.datasource_uid,
                "expr": "1",
                "intervalMs": 30000,
                "maxDataPoints": 10,
            }
        ],
        "from": "now-5m",
        "to": "now",
    }
    request = urllib.request.Request(
        url,
        method="POST",
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {runtime.grafana_token}",
            "Content-Type": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=8) as response:
            _ = json.loads(response.read().decode("utf-8"))
            return True, "grafana query api reachable"
    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, json.JSONDecodeError) as exc:
        return False, f"grafana query api unreachable: {exc}"


def run_connectivity_checks(datasources: Iterable[str], runtime: RuntimeConfig) -> List[ConnectivityCheck]:
    checks: List[ConnectivityCheck] = []

    normalized = sorted({(name or "prometheus").lower().replace(" ", "_") for name in datasources})
    if not normalized:
        normalized = ["prometheus"]

    for datasource in normalized:
        exists = datasource in KNOWN_DATASOURCES

        if runtime.offline:
            credentials_ok = True
            query_api_ok = True
            message = "offline mode: runtime api checks skipped"
        elif datasource == "prometheus":
            credentials_ok = bool(runtime.prom_url or (runtime.grafana_url and runtime.grafana_token and runtime.datasource_uid))
            if runtime.prom_url:
                query_api_ok, message = _check_prometheus_query_api(runtime)
            else:
                query_api_ok, message = _check_grafana_query_api(runtime)
        else:
            credentials_ok = bool(runtime.grafana_url and runtime.grafana_token and runtime.datasource_uid)
            query_api_ok, message = _check_grafana_query_api(runtime)

        if not exists:
            status = "failed"
            message = f"unknown datasource: {datasource}"
        elif not credentials_ok:
            status = "failed"
            message = "credentials unavailable"
        elif not query_api_ok:
            status = "failed"
        else:
            status = "ok"

        checks.append(
            ConnectivityCheck(
                datasource=datasource,
                exists=exists,
                credentials_ok=credentials_ok,
                query_api_ok=query_api_ok,
                status=status,
                message=message,
            )
        )

    return checks

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

from __future__ import annotations

import json
from dataclasses import asdict
from pathlib import Path
from typing import Dict

from .models import PipelineArtifacts


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 "dashboard"


def write_outputs(out_base_dir: str, repo_slug: str, artifacts: PipelineArtifacts) -> str:
    outdir = Path(out_base_dir) / _slug(repo_slug)
    outdir.mkdir(parents=True, exist_ok=True)

    (outdir / "dashboard-assembled.json").write_text(
        json.dumps(artifacts.dashboard_assembled, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    validation_report: Dict[str, object] = {
        "pipeline_validation": artifacts.validation.to_dict(),
        "connectivity_checks": [asdict(item) for item in artifacts.connectivity_checks],
        "target_validations": [asdict(item) for item in artifacts.target_validations],
        "panel_findings": [asdict(item) for item in artifacts.panel_findings],
    }
    (outdir / "validation-report.json").write_text(
        json.dumps(validation_report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    (outdir / "autofix-report.json").write_text(
        json.dumps({"fix_actions": [asdict(item) for item in artifacts.fix_actions]}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    (outdir / "acceptance-report.json").write_text(
        json.dumps(artifacts.acceptance.to_dict(), ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    (outdir / "traceability.json").write_text(
        json.dumps(artifacts.traceability, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    (outdir / "evidence-index.enriched.json").write_text(
        json.dumps(artifacts.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

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

from .adapters import (
    load_existing_dashboard,
    load_log_dict,
    load_metric_mapping_spec,
    load_metrics_catalog,
    load_sli_spec,
    load_trace_spans,
)
from .models import (
    CatalogMetric,
    EvidenceItem,
    MetricMappingItem,
    NormalizedAssemblyInputs,
    PanelTarget,
    RuntimeConfig,
    SLIItem,
)


TOKEN_PATTERN = re.compile(r"[a-z0-9_]+")


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 _to_dimensions(value: Any) -> List[str]:
    if isinstance(value, list):
        return [str(item).strip() for item in value if str(item).strip()]
    if isinstance(value, str):
        return [item.strip() for item in value.split(",") if item.strip()]
    return []


def _normalize_sli_items(payload: Any) -> List[SLIItem]:
    records: List[Dict[str, Any]] = []
    if isinstance(payload, list):
        records.extend(item for item in payload if isinstance(item, dict))
    elif isinstance(payload, dict):
        for key in ["sli_indicators", "indicators", "slis", "items"]:
            value = payload.get(key)
            if isinstance(value, list):
                records.extend(item for item in value if isinstance(item, dict))
        if not records:
            records.append(payload)

    result: List[SLIItem] = []
    for index, item in enumerate(records, start=1):
        result.append(
            SLIItem(
                sli_name=_pick_text(item.get("sli_name"), item.get("name"), item.get("id"), default=f"sli-{index}"),
                sli_type=_pick_text(item.get("sli_type"), item.get("type"), default="availability"),
                measurement=_pick_text(item.get("measurement"), item.get("formula"), item.get("query")),
                target=_pick_text(item.get("target"), item.get("target_slo"), item.get("objective"), item.get("slo")),
                dimensions=_to_dimensions(item.get("dimensions") or item.get("dimension")),
            )
        )
    return result


def _normalize_mapping_items(payload: Any) -> List[MetricMappingItem]:
    records = payload if isinstance(payload, list) else _as_list(payload.get("items") if isinstance(payload, dict) else payload)
    result: List[MetricMappingItem] = []
    for item in records:
        if not isinstance(item, dict):
            continue
        datasource = _pick_text(item.get("datasource"), default="prometheus").lower().replace(" ", "_")
        if datasource in {"internal", "internal-tsdb", "internaltsdb"}:
            datasource = "internal_tsdb"
        result.append(
            MetricMappingItem(
                sli_name=_pick_text(item.get("sli_name"), item.get("name")),
                chosen_metric=_pick_text(item.get("chosen_metric"), item.get("metric")),
                datasource=datasource,
                query_template=_pick_text(item.get("query_template"), item.get("query")),
                dimensions=_to_dimensions(item.get("dimensions") or item.get("labels")),
                confidence=float(item.get("confidence") or 0.0),
                missing_gap=_pick_text(item.get("missing_gap"), default="none"),
            )
        )
    return result


def _normalize_catalog_metrics(payload: Any) -> List[CatalogMetric]:
    records: List[Dict[str, Any]] = []
    if isinstance(payload, list):
        records.extend(item for item in payload if isinstance(item, dict))
    elif isinstance(payload, dict):
        values = payload.get("metrics") or payload.get("items")
        if isinstance(values, list):
            records.extend(item for item in values if isinstance(item, dict))
        elif payload:
            records.append(payload)

    result: List[CatalogMetric] = []
    for item in records:
        name = _pick_text(item.get("name"), item.get("metric"), item.get("metric_name"))
        if not name:
            continue
        datasource = _pick_text(item.get("datasource"), item.get("source"), default="prometheus").lower().replace(" ", "_")
        if datasource in {"internal", "internal-tsdb", "internaltsdb"}:
            datasource = "internal_tsdb"
        result.append(CatalogMetric(name=name, datasource=datasource, dimensions=_to_dimensions(item.get("dimensions") or item.get("labels"))))
    return result


def _iter_panels(panels: Iterable[Dict[str, Any]]) -> Iterable[Dict[str, Any]]:
    for panel in panels:
        yield panel
        nested = panel.get("panels")
        if isinstance(nested, list):
            for sub in _iter_panels(nested):
                yield sub


def _extract_panel_targets(dashboard: Dict[str, Any], fallback_datasource: str = "prometheus") -> List[PanelTarget]:
    result: List[PanelTarget] = []
    for panel in _iter_panels(dashboard.get("panels", [])):
        panel_id = int(panel.get("id") or -1)
        panel_title = _pick_text(panel.get("title"), default="Untitled")
        panel_type = _pick_text(panel.get("type"), default="timeseries")
        panel_ds = panel.get("datasource")
        panel_ds_name = fallback_datasource
        if isinstance(panel_ds, dict):
            panel_ds_name = _pick_text(panel_ds.get("type"), panel_ds.get("uid"), default=fallback_datasource)
        elif isinstance(panel_ds, str) and panel_ds.strip():
            panel_ds_name = panel_ds.strip()

        targets = panel.get("targets") if isinstance(panel.get("targets"), list) else []
        for index, target in enumerate(targets):
            if not isinstance(target, dict):
                continue
            query = _pick_text(target.get("expr"), target.get("query"))
            ref_id = _pick_text(target.get("refId"), default=chr(ord("A") + index))
            target_ds = target.get("datasource")
            datasource = panel_ds_name
            if isinstance(target_ds, dict):
                datasource = _pick_text(target_ds.get("type"), target_ds.get("uid"), default=datasource)
            elif isinstance(target_ds, str) and target_ds.strip():
                datasource = target_ds.strip()

            result.append(
                PanelTarget(
                    panel_id=panel_id,
                    panel_title=panel_title,
                    panel_type=panel_type,
                    target_index=index,
                    ref_id=ref_id,
                    datasource=datasource.lower().replace(" ", "_"),
                    query=query,
                )
            )
    return result


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 "dashboard"


def _derive_repo_slug(existing_dashboard_path: str, dashboard: Dict[str, Any], focus_service: str | None) -> str:
    if focus_service and focus_service.strip():
        return _slug(focus_service)
    title = _pick_text(dashboard.get("title"))
    if title:
        return _slug(title)
    return _slug(Path(existing_dashboard_path).stem)


def normalize_inputs(
    sli_spec_path: str,
    metric_mapping_spec_path: str,
    metrics_catalog_path: str,
    log_dict_path: str,
    trace_spans_path: str,
    existing_dashboard_path: str,
    runtime: RuntimeConfig,
    focus_service: str | None = None,
) -> NormalizedAssemblyInputs:
    sli_payload = load_sli_spec(sli_spec_path)
    mapping_payload = load_metric_mapping_spec(metric_mapping_spec_path)
    catalog_payload = load_metrics_catalog(metrics_catalog_path)
    log_payload = load_log_dict(log_dict_path)
    trace_payload = load_trace_spans(trace_spans_path)
    dashboard_payload = load_existing_dashboard(existing_dashboard_path)

    sli_items = _normalize_sli_items(sli_payload)
    mapping_items = _normalize_mapping_items(mapping_payload)
    catalog_metrics = _normalize_catalog_metrics(catalog_payload)

    log_fields = [str(v).strip() for v in _as_list(log_payload.get("fields") if isinstance(log_payload, dict) else log_payload) if str(v).strip()]
    trace_spans = [str(v).strip() for v in _as_list(trace_payload.get("spans") if isinstance(trace_payload, dict) else trace_payload) if str(v).strip()]

    dashboard = dashboard_payload if isinstance(dashboard_payload, dict) else {}
    panel_targets = _extract_panel_targets(dashboard)

    evidence_items: List[EvidenceItem] = []
    for idx, item in enumerate(sli_items, start=1):
        evidence_items.append(
            EvidenceItem(
                evidence_id=f"ev-sli-{idx}",
                source_type="sli_spec",
                source_path=sli_spec_path,
                locator=f"sli[{idx-1}]",
                summary=item.sli_name,
            )
        )
    for idx, item in enumerate(mapping_items, start=1):
        evidence_items.append(
            EvidenceItem(
                evidence_id=f"ev-map-{idx}",
                source_type="metric_mapping_spec",
                source_path=metric_mapping_spec_path,
                locator=f"mapping[{idx-1}]",
                summary=f"{item.sli_name} -> {item.chosen_metric}",
            )
        )
    for idx, item in enumerate(catalog_metrics, start=1):
        evidence_items.append(
            EvidenceItem(
                evidence_id=f"ev-catalog-{idx}",
                source_type="metrics_catalog",
                source_path=metrics_catalog_path,
                locator=item.name,
                summary=item.name,
            )
        )
    for idx, field in enumerate(log_fields, start=1):
        evidence_items.append(
            EvidenceItem(
                evidence_id=f"ev-log-{idx}",
                source_type="log_dict",
                source_path=log_dict_path,
                locator=field,
                summary=field,
            )
        )
    for idx, span in enumerate(trace_spans, start=1):
        evidence_items.append(
            EvidenceItem(
                evidence_id=f"ev-trace-{idx}",
                source_type="trace_spans",
                source_path=trace_spans_path,
                locator=span,
                summary=span,
            )
        )

    return NormalizedAssemblyInputs(
        repo_slug=_derive_repo_slug(existing_dashboard_path, dashboard, focus_service),
        sli_items=sli_items,
        mapping_items=mapping_items,
        catalog_metrics=catalog_metrics,
        log_fields=log_fields,
        trace_spans=trace_spans,
        dashboard=dashboard,
        panel_targets=panel_targets,
        evidence_items=evidence_items,
        runtime=runtime,
    )

# 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 SLIItem:
    sli_name: str
    sli_type: str
    measurement: str = ""
    target: str = ""
    dimensions: List[str] = field(default_factory=list)


@dataclass
class MetricMappingItem:
    sli_name: str
    chosen_metric: str
    datasource: str
    query_template: str
    dimensions: List[str] = field(default_factory=list)
    confidence: float = 0.0
    missing_gap: str = "none"


@dataclass
class CatalogMetric:
    name: str
    datasource: str
    dimensions: List[str] = field(default_factory=list)


@dataclass
class PanelTarget:
    panel_id: int
    panel_title: str
    panel_type: str
    target_index: int
    ref_id: str
    datasource: str
    query: str


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


@dataclass
class RuntimeConfig:
    offline: bool = False
    grafana_url: str = ""
    grafana_token: str = ""
    datasource_uid: str = ""
    prom_url: str = ""
    prom_bearer: str = ""
    prom_username: str = ""
    prom_password: str = ""
    time_range: str = "now-6h,now"


@dataclass
class NormalizedAssemblyInputs:
    repo_slug: str
    sli_items: List[SLIItem] = field(default_factory=list)
    mapping_items: List[MetricMappingItem] = field(default_factory=list)
    catalog_metrics: List[CatalogMetric] = field(default_factory=list)
    log_fields: List[str] = field(default_factory=list)
    trace_spans: List[str] = field(default_factory=list)
    dashboard: Dict[str, Any] = field(default_factory=dict)
    panel_targets: List[PanelTarget] = field(default_factory=list)
    evidence_items: List[EvidenceItem] = field(default_factory=list)
    runtime: RuntimeConfig = field(default_factory=RuntimeConfig)


@dataclass
class ConnectivityCheck:
    datasource: str
    exists: bool
    credentials_ok: bool
    query_api_ok: bool
    status: str
    message: str


@dataclass
class TargetValidation:
    panel_id: int
    panel_title: str
    target_index: int
    ref_id: str
    datasource: str
    query: str
    executable: bool
    labels_ok: bool
    aggregation_ok: bool
    has_data: bool
    status: str
    errors: List[str] = field(default_factory=list)


@dataclass
class PanelFinding:
    panel_id: int
    panel_title: str
    severity: str
    finding_type: str
    message: str
    target_index: Optional[int] = None


@dataclass
class FixAction:
    panel_id: int
    panel_title: str
    target_index: int
    action: str
    before: str
    after: str
    reason: str
    confidence: float


@dataclass
class AcceptanceReport:
    success_rate: float
    empty_panels_count: int
    error_query_count: int
    manual_confirmation_items: List[str]
    overall_pass: bool

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


@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(item) for item in self.issues],
        }


@dataclass
class PipelineArtifacts:
    dashboard_assembled: Dict[str, Any]
    connectivity_checks: List[ConnectivityCheck]
    target_validations: List[TargetValidation]
    panel_findings: List[PanelFinding]
    fix_actions: List[FixAction]
    acceptance: AcceptanceReport
    validation: ValidationReport
    traceability: Dict[str, Any]
    evidence_enriched: List[Dict[str, Any]]


@dataclass
class PipelineResult:
    output_dir: str
    panel_count: int
    target_count: int
    overall_pass: bool

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

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

from __future__ import annotations

import re
from typing import Dict, List

from .models import PanelFinding, TargetValidation


def _grouping_label_count(query: str) -> int:
    total = 0
    for match in re.finditer(r"by\s*\(([^)]*)\)", query):
        parts = [item.strip() for item in match.group(1).split(",") if item.strip()]
        total += len(parts)
    return total


def detect_panel_quality(validations: List[TargetValidation]) -> List[PanelFinding]:
    findings: List[PanelFinding] = []

    for item in validations:
        if not item.has_data:
            findings.append(
                PanelFinding(
                    panel_id=item.panel_id,
                    panel_title=item.panel_title,
                    severity="warning",
                    finding_type="no_data",
                    message="query has no data in selected range",
                    target_index=item.target_index,
                )
            )
        if not item.labels_ok:
            findings.append(
                PanelFinding(
                    panel_id=item.panel_id,
                    panel_title=item.panel_title,
                    severity="warning",
                    finding_type="wrong_labels",
                    message="query labels do not satisfy required dimensions",
                    target_index=item.target_index,
                )
            )
        if not item.aggregation_ok:
            findings.append(
                PanelFinding(
                    panel_id=item.panel_id,
                    panel_title=item.panel_title,
                    severity="warning",
                    finding_type="misleading_aggregation",
                    message="aggregation semantics do not match sli intent",
                    target_index=item.target_index,
                )
            )
        if item.status == "error":
            findings.append(
                PanelFinding(
                    panel_id=item.panel_id,
                    panel_title=item.panel_title,
                    severity="blocker",
                    finding_type="query_error",
                    message="query is not executable",
                    target_index=item.target_index,
                )
            )

        grouping = _grouping_label_count(item.query)
        if grouping >= 4:
            findings.append(
                PanelFinding(
                    panel_id=item.panel_id,
                    panel_title=item.panel_title,
                    severity="warning",
                    finding_type="result_density_risk",
                    message="query grouping is too dense and may create noisy panel output",
                    target_index=item.target_index,
                )
            )

    by_panel: Dict[int, int] = {}
    for finding in findings:
        by_panel[finding.panel_id] = by_panel.get(finding.panel_id, 0) + 1

    for panel_id, count in by_panel.items():
        if count >= 3:
            title = next((item.panel_title for item in validations if item.panel_id == panel_id), "panel")
            findings.append(
                PanelFinding(
                    panel_id=panel_id,
                    panel_title=title,
                    severity="warning",
                    finding_type="panel_quality_risk",
                    message="panel has multiple quality findings and needs manual confirmation",
                )
            )

    return findings

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

from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

from .acceptance import build_acceptance_report
from .auto_repair import apply_auto_repair
from .connectivity_checker import run_connectivity_checks
from .exporter import write_outputs
from .input_normalizer import normalize_inputs
from .models import PipelineArtifacts, PipelineResult, RuntimeConfig
from .panel_quality_detector import detect_panel_quality
from .query_adapter import adapt_queries
from .traceability import build_traceability
from .validator import validate_pipeline


@dataclass
class PipelineOptions:
    out_dir: str = "output"
    focus_service: Optional[str] = None
    offline: bool = False
    grafana_url: str = ""
    grafana_token: str = ""
    datasource_uid: str = ""
    prom_url: str = ""
    prom_bearer: str = ""
    prom_username: str = ""
    prom_password: str = ""
    time_range: str = "now-6h,now"
    max_repair_rounds: int = 2


def run_pipeline(
    sli_spec: str,
    metric_mapping_spec: str,
    metrics_catalog: str,
    log_dict: str,
    trace_spans: str,
    existing_dashboard: str,
    options: PipelineOptions | None = None,
) -> PipelineResult:
    opts = options or PipelineOptions()
    runtime = RuntimeConfig(
        offline=opts.offline,
        grafana_url=opts.grafana_url,
        grafana_token=opts.grafana_token,
        datasource_uid=opts.datasource_uid,
        prom_url=opts.prom_url,
        prom_bearer=opts.prom_bearer,
        prom_username=opts.prom_username,
        prom_password=opts.prom_password,
        time_range=opts.time_range,
    )

    normalized = normalize_inputs(
        sli_spec_path=sli_spec,
        metric_mapping_spec_path=metric_mapping_spec,
        metrics_catalog_path=metrics_catalog,
        log_dict_path=log_dict,
        trace_spans_path=trace_spans,
        existing_dashboard_path=existing_dashboard,
        runtime=runtime,
        focus_service=opts.focus_service,
    )

    datasource_names = [item.datasource for item in normalized.mapping_items] + [item.datasource for item in normalized.panel_targets]
    connectivity_checks = run_connectivity_checks(datasource_names, normalized.runtime)

    dashboard = normalized.dashboard
    dashboard_adapted, target_validations = adapt_queries(normalized, dashboard)
    panel_findings = detect_panel_quality(target_validations)

    fix_actions = []
    current_dashboard = dashboard_adapted
    current_validations = target_validations
    current_findings = panel_findings

    for _ in range(max(1, opts.max_repair_rounds)):
        has_issue = any(item.status != "success" for item in current_validations)
        if not has_issue:
            break

        repaired_dashboard, round_actions = apply_auto_repair(
            inputs=normalized,
            dashboard=current_dashboard,
            validations=current_validations,
            findings=current_findings,
        )
        if not round_actions:
            break

        fix_actions.extend(round_actions)
        current_dashboard, current_validations = adapt_queries(normalized, repaired_dashboard)
        current_findings = detect_panel_quality(current_validations)

    acceptance = build_acceptance_report(current_validations, current_findings)
    pipeline_validation = validate_pipeline(connectivity_checks, current_validations, acceptance)
    traceability = build_traceability(
        mappings=normalized.mapping_items,
        validations=current_validations,
        findings=current_findings,
        fixes=fix_actions,
        evidence_items=normalized.evidence_items,
    )

    evidence_enriched = [
        {
            "evidence_id": item.evidence_id,
            "source_type": item.source_type,
            "source_path": item.source_path,
            "locator": item.locator,
            "summary": item.summary,
        }
        for item in normalized.evidence_items
    ]

    artifacts = PipelineArtifacts(
        dashboard_assembled=current_dashboard,
        connectivity_checks=connectivity_checks,
        target_validations=current_validations,
        panel_findings=current_findings,
        fix_actions=fix_actions,
        acceptance=acceptance,
        validation=pipeline_validation,
        traceability=traceability,
        evidence_enriched=evidence_enriched,
    )

    output_dir = write_outputs(opts.out_dir, normalized.repo_slug, artifacts)
    return PipelineResult(
        output_dir=output_dir,
        panel_count=len(current_dashboard.get("panels", [])),
        target_count=len(current_validations),
        overall_pass=acceptance.overall_pass and pipeline_validation.passed,
    )

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

from __future__ import annotations

import re
from copy import deepcopy
from typing import Dict, List, Tuple

from .models import MetricMappingItem, NormalizedAssemblyInputs, PanelTarget, TargetValidation


TOKEN_PATTERN = re.compile(r"[a-z0-9_]+")
METRIC_PATTERN = re.compile(r"\b([a-zA-Z_:][a-zA-Z0-9_:]*)\b")


def _tokens(text: str) -> set[str]:
    return {token for token in TOKEN_PATTERN.findall(text.lower()) if len(token) >= 3}


def _balanced(text: str, left: str, right: str) -> bool:
    count = 0
    for ch in text:
        if ch == left:
            count += 1
        elif ch == right:
            count -= 1
            if count < 0:
                return False
    return count == 0


def _query_executable(query: str) -> tuple[bool, List[str]]:
    errors: List[str] = []
    if not query.strip():
        errors.append("empty query")
    if not _balanced(query, "(", ")"):
        errors.append("unbalanced parentheses")
    if not _balanced(query, "{", "}"):
        errors.append("unbalanced braces")
    return len(errors) == 0, errors


def _extract_labels(query: str) -> set[str]:
    labels: set[str] = set()
    match = re.search(r"\{([^}]*)\}", query)
    if match:
        for part in match.group(1).split(","):
            key = re.split(r"=~|!=|=", part.strip(), maxsplit=1)[0].strip()
            if key:
                labels.add(key)

    for match in re.finditer(r"by\s*\(([^)]*)\)", query):
        for part in match.group(1).split(","):
            key = part.strip()
            if key:
                labels.add(key)

    return labels


def _extract_metric_names(query: str) -> set[str]:
    reserved = {
        "sum",
        "avg",
        "min",
        "max",
        "count",
        "rate",
        "irate",
        "increase",
        "histogram_quantile",
        "topk",
        "bottomk",
        "clamp_min",
        "clamp_max",
        "by",
        "without",
    }
    names = set()
    for token in METRIC_PATTERN.findall(query):
        if token not in reserved:
            names.add(token)
    return names


def _catalog_dimensions_for_query(query: str, catalog_dimensions: Dict[str, set[str]]) -> set[str]:
    dimensions: set[str] = set()
    for metric_name in _extract_metric_names(query):
        dimensions.update(catalog_dimensions.get(metric_name, set()))
    return dimensions


def _aggregation_ok(query: str, sli_type: str) -> bool:
    stype = sli_type.lower()
    lower = query.lower()
    if "latency" in stype:
        return any(token in lower for token in ["histogram_quantile", "quantile", "p95", "p99"])
    if any(token in stype for token in ["availability", "correctness", "completeness"]):
        return "/" in lower or "ratio" in lower
    return True


def _guess_has_data(query: str) -> bool:
    lower = query.lower()
    if any(token in lower for token in ["absent(", "no_data_metric", "missing_metric"]):
        return False
    return True


def _best_mapping(target: PanelTarget, mapping_items: List[MetricMappingItem]) -> MetricMappingItem | None:
    title_tokens = _tokens(target.panel_title)
    query_tokens = _tokens(target.query)

    scored: List[tuple[MetricMappingItem, float]] = []
    for item in mapping_items:
        score = 0.0
        sli_tokens = _tokens(item.sli_name)
        score += len(title_tokens & sli_tokens) * 0.3
        score += len(query_tokens & sli_tokens) * 0.2
        if item.chosen_metric and item.chosen_metric in target.query:
            score += 0.8
        if item.datasource and target.datasource.startswith(item.datasource):
            score += 0.2
        scored.append((item, score))

    scored.sort(key=lambda pair: pair[1], reverse=True)
    if not scored:
        return None
    if scored[0][1] <= 0:
        return None
    return scored[0][0]


def _normalize_datasource_type(name: str) -> str:
    normalized = (name or "prometheus").strip().lower().replace(" ", "_")
    return normalized or "prometheus"


def _datasource_var_name(datasource_type: str) -> str:
    return "DS_" + _normalize_datasource_type(datasource_type).upper()


def _grafana_datasource_ref(datasource_type: str, runtime_uid: str) -> Dict[str, str]:
    normalized = _normalize_datasource_type(datasource_type)
    if runtime_uid and normalized == "prometheus":
        return {"type": normalized, "uid": runtime_uid}
    return {"type": normalized, "uid": "${" + _datasource_var_name(normalized) + "}"}


def _uses_datasource_variable(datasource_type: str, runtime_uid: str) -> bool:
    normalized = _normalize_datasource_type(datasource_type)
    return not (runtime_uid and normalized == "prometheus")


def _plugin_name(datasource_type: str) -> str:
    normalized = _normalize_datasource_type(datasource_type)
    if normalized == "prometheus":
        return "Prometheus"
    if normalized == "tempo":
        return "Tempo"
    if normalized == "loki":
        return "Loki"
    return normalized


def _datasource_input(datasource_type: str) -> Dict[str, str]:
    normalized = _normalize_datasource_type(datasource_type)
    return {
        "name": _datasource_var_name(normalized),
        "label": _plugin_name(normalized),
        "description": "",
        "type": "datasource",
        "pluginId": normalized,
        "pluginName": _plugin_name(normalized),
    }


def _merge_dashboard_inputs(existing: object, datasource_types: set[str]) -> List[Dict[str, str]]:
    merged: Dict[str, Dict[str, str]] = {}
    if isinstance(existing, list):
        for item in existing:
            if isinstance(item, dict) and isinstance(item.get("name"), str):
                merged[item["name"]] = item

    for datasource_type in sorted(datasource_types):
        item = _datasource_input(datasource_type)
        merged[item["name"]] = item

    return [merged[name] for name in sorted(merged)]


def _require_item(datasource_type: str) -> Dict[str, str]:
    normalized = _normalize_datasource_type(datasource_type)
    return {
        "type": "datasource",
        "id": normalized,
        "name": _plugin_name(normalized),
        "version": "",
    }


def _merge_requires(existing: object, datasource_types: set[str]) -> List[Dict[str, str]]:
    merged: Dict[str, Dict[str, str]] = {}
    if isinstance(existing, list):
        for item in existing:
            if isinstance(item, dict) and isinstance(item.get("id"), str):
                merged[item["id"]] = item

    for datasource_type in sorted(datasource_types):
        item = _require_item(datasource_type)
        merged[item["id"]] = item

    return [merged[name] for name in sorted(merged)]


def adapt_queries(inputs: NormalizedAssemblyInputs, dashboard: Dict[str, object]) -> Tuple[Dict[str, object], List[TargetValidation]]:
    patched = deepcopy(dashboard)
    panel_targets = inputs.panel_targets

    target_lookup = {(item.panel_id, item.target_index): item for item in panel_targets}
    mapping_by_target: Dict[tuple[int, int], MetricMappingItem | None] = {}
    for target in panel_targets:
        mapping_by_target[(target.panel_id, target.target_index)] = _best_mapping(target, inputs.mapping_items)

    catalog_dimensions: Dict[str, set[str]] = {item.name: set(item.dimensions) for item in inputs.catalog_metrics}

    validations: List[TargetValidation] = []
    datasource_inputs: set[str] = set()

    for panel in patched.get("panels", []):
        targets = panel.get("targets") if isinstance(panel.get("targets"), list) else []
        for index, target in enumerate(targets):
            if not isinstance(target, dict):
                continue
            panel_id = int(panel.get("id") or -1)
            key = (panel_id, index)
            base = target_lookup.get(key)
            if base is None:
                continue

            mapping = mapping_by_target.get(key)
            query = (target.get("expr") or target.get("query") or "").strip()
            if not query and mapping and mapping.query_template:
                query = mapping.query_template

            if mapping and mapping.chosen_metric and mapping.chosen_metric not in query and mapping.query_template:
                query = mapping.query_template

            if "expr" in target:
                target["expr"] = query
            else:
                target["query"] = query

            source_ds = mapping.datasource if mapping else base.datasource
            datasource_ref = _grafana_datasource_ref(source_ds, inputs.runtime.datasource_uid)
            if _uses_datasource_variable(source_ds, inputs.runtime.datasource_uid):
                datasource_inputs.add(_normalize_datasource_type(source_ds))
            target["datasource"] = datasource_ref
            panel["datasource"] = datasource_ref

            executable, errors = _query_executable(query)
            required_labels = set(mapping.dimensions) if mapping else set()
            existing_labels = _extract_labels(query)
            available_from_catalog = _catalog_dimensions_for_query(query, catalog_dimensions)
            labels_ok = required_labels.issubset(existing_labels | available_from_catalog) if required_labels else True

            sli_type = mapping.sli_name if mapping else panel.get("title", "")
            aggregation_ok = _aggregation_ok(query, sli_type)
            has_data = _guess_has_data(query)

            if not executable:
                status = "error"
            elif executable and labels_ok and aggregation_ok and has_data:
                status = "success"
            else:
                status = "warning"

            if not labels_ok:
                missing = sorted(required_labels - existing_labels)
                errors.append(f"missing labels: {', '.join(missing)}")
            if not aggregation_ok:
                errors.append("aggregation does not match sli semantics")
            if not has_data:
                errors.append("query has no data in selected range")

            validations.append(
                TargetValidation(
                    panel_id=panel_id,
                    panel_title=str(panel.get("title") or "Untitled"),
                    target_index=index,
                    ref_id=str(target.get("refId") or chr(ord("A") + index)),
                    datasource=base.datasource,
                    query=query,
                    executable=executable,
                    labels_ok=labels_ok,
                    aggregation_ok=aggregation_ok,
                    has_data=has_data,
                    status=status,
                    errors=errors,
                )
            )

    if datasource_inputs:
        patched["__inputs"] = _merge_dashboard_inputs(patched.get("__inputs"), datasource_inputs)
        patched["__requires"] = _merge_requires(patched.get("__requires"), datasource_inputs)

    return patched, validations

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

from __future__ import annotations

from typing import Any, Dict, List

from .models import EvidenceItem, FixAction, MetricMappingItem, PanelFinding, TargetValidation


def build_traceability(
    mappings: List[MetricMappingItem],
    validations: List[TargetValidation],
    findings: List[PanelFinding],
    fixes: List[FixAction],
    evidence_items: List[EvidenceItem],
) -> Dict[str, Any]:
    links: List[Dict[str, Any]] = []

    for item in validations:
        matched = [
            ev.evidence_id
            for ev in evidence_items
            if item.panel_title.lower() in ev.summary.lower() or any(token in ev.summary for token in [item.ref_id, str(item.panel_id)])
        ]
        if not matched and evidence_items:
            matched = [evidence_items[0].evidence_id]

        links.append(
            {
                "target_type": "panel_query",
                "target_id": f"{item.panel_id}:{item.target_index}",
                "evidence_ids": matched,
                "rationale": f"validated query for panel={item.panel_title} status={item.status}",
            }
        )

    for fix in fixes:
        links.append(
            {
                "target_type": "auto_fix",
                "target_id": f"{fix.panel_id}:{fix.target_index}:{fix.action}",
                "evidence_ids": [ev.evidence_id for ev in evidence_items[:3]],
                "rationale": f"applied {fix.action} due to {fix.reason}",
            }
        )

    for finding in findings:
        links.append(
            {
                "target_type": "panel_finding",
                "target_id": f"{finding.panel_id}:{finding.finding_type}",
                "evidence_ids": [ev.evidence_id for ev in evidence_items[:2]],
                "rationale": finding.message,
            }
        )

    return {
        "trace_links": links,
        "mapping_summary": [
            {
                "sli_name": item.sli_name,
                "chosen_metric": item.chosen_metric,
                "datasource": item.datasource,
                "confidence": item.confidence,
            }
            for item in mappings
        ],
        "evidence_items": [
            {
                "evidence_id": ev.evidence_id,
                "source_type": ev.source_type,
                "source_path": ev.source_path,
                "locator": ev.locator,
                "summary": ev.summary,
            }
            for ev in evidence_items
        ],
    }