Byted Sol Stability Sli Modeling

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-sli-modeling"
version = "0.1.0"
description = "Generate structured SLI specs for SLO and error budget management"
readme = "SKILL.md"
requires-python = ">=3.10"
dependencies = []

[project.scripts]
byted-sol-stablity-sli-modeling = "sli_modeling_skill.cli:main"

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

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

from .models import SLISpec, SLIType, Severity
from .modeler import build_sli_specs

__all__ = ["SLISpec", "SLIType", "Severity", "build_sli_specs"]

# 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 .exporter import write_outputs
from .modeler import build_sli_specs


def _parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="Build structured SLI specs")
    p.add_argument("--input", required=True, help="Path to capability text file")
    p.add_argument("--owner", required=True, help="Owner of the SLI")
    p.add_argument("--reference", action="append", default=[], help="Optional reference files")
    p.add_argument("--out-dir", default="output")
    return p


def run_cli(argv: List[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    input_path = Path(args.input)
    if not input_path.exists() or input_path.is_dir():
        raise SystemExit(f"input file not found: {input_path}")

    text = input_path.read_text(encoding="utf-8")
    result = build_sli_specs(text, owner=args.owner, reference_paths=args.reference)
    output_dir = write_outputs(result, out_base_dir=args.out_dir)

    report = {
        "output_dir": output_dir,
        "total_specs": len(result.specs),
        "notes": result.notes,
    }
    print(json.dumps(report, 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 pathlib import Path
import json
from typing import List

from .modeler import ModelerResult


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 "sli-spec"


def write_outputs(result: ModelerResult, out_base_dir: str) -> str:
    if not result.specs:
        raise ValueError("no sli specs to export")

    slug = _slug(result.specs[0].capability)
    outdir = Path(out_base_dir) / slug
    outdir.mkdir(parents=True, exist_ok=True)

    payload = [x.to_dict() for x in result.specs]
    (outdir / "sli-spec.json").write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    report_lines: List[str] = ["# SLI Modeling Report", ""]
    for idx, spec in enumerate(payload, 1):
        report_lines.append(f"## Spec {idx}")
        for k, v in spec.items():
            report_lines.append(f"- {k}: `{v}`")
        report_lines.append("")
    if result.notes:
        report_lines.append("## Notes")
        for n in result.notes:
            report_lines.append(f"- {n}")

    (outdir / "sli-report.md").write_text("\n".join(report_lines), encoding="utf-8")
    return str(outdir)

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

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
import re
from typing import Dict, Iterable, List

from .models import SLISpec, SLIType, Severity


@dataclass
class ModelerResult:
    specs: List[SLISpec]
    notes: List[str]


def _extract_key_values(text: str) -> Dict[str, str]:
    result: Dict[str, str] = {}
    for line in text.splitlines():
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        key = key.strip().lower().replace("-", "_").replace(" ", "_")
        value = value.strip()
        if key and value:
            result[key] = value
    return result


def _infer_sli_type(text: str) -> SLIType:
    low = text.lower()
    if any(x in low for x in ["latency", "p95", "p99", "duration", "耗时", "延迟"]):
        return SLIType.LATENCY
    if any(x in low for x in ["correct", "accuracy", "正确", "错误率", "error rate"]):
        return SLIType.CORRECTNESS
    if any(x in low for x in ["fresh", "时效", "新鲜", "lag"]):
        return SLIType.FRESHNESS
    if any(x in low for x in ["complete", "coverage", "完整", "丢失"]):
        return SLIType.COMPLETENESS
    if any(x in low for x in ["consistent", "一致", "一致性"]):
        return SLIType.CONSISTENCY
    return SLIType.AVAILABILITY


def _looks_like_internal_resource_metric(text: str) -> bool:
    low = text.lower()
    resource_hints = ("cpu", "ram", "storage", "throughput", "loadavg")
    return any(x in low for x in resource_hints)


def _infer_severity(text: str) -> Severity:
    m = re.search(r"\bP([012])\b", text.upper())
    if not m:
        return Severity.P1
    return {"0": Severity.P0, "1": Severity.P1, "2": Severity.P2}[m.group(1)]


def _read_refs(paths: Iterable[str]) -> str:
    chunks: List[str] = []
    for raw in paths:
        p = Path(raw)
        if not p.exists() or p.is_dir():
            continue
        try:
            chunks.append(p.read_text(encoding="utf-8", errors="ignore"))
        except OSError:
            continue
    return "\n".join(chunks)


def build_sli_specs(input_text: str, owner: str, reference_paths: Iterable[str] = ()) -> ModelerResult:
    if not input_text.strip():
        raise ValueError("input text cannot be empty")
    if not owner.strip():
        raise ValueError("owner cannot be empty")

    ref_text = _read_refs(reference_paths)
    merged = "\n".join([input_text, ref_text])
    kv = _extract_key_values(input_text)

    capability = kv.get("capability") or input_text.strip().splitlines()[0].strip()[:120]
    user_journey = kv.get("user_journey") or "service request handling"
    sli_type = kv.get("sli_type") or _infer_sli_type(merged).value
    severity = kv.get("severity") or _infer_severity(merged).value

    sli_name = kv.get("sli_name") or f"{capability} {sli_type}"
    measurement = kv.get("measurement")
    denominator = kv.get("denominator")

    if not measurement:
        if sli_type == SLIType.LATENCY.value:
            measurement = "good_requests_under_300ms / total_requests"
        elif sli_type == SLIType.CORRECTNESS.value:
            measurement = "correct_responses / processed_requests"
        elif sli_type == SLIType.FRESHNESS.value:
            measurement = "records_within_freshness_window / expected_records"
        else:
            measurement = "successful_requests / total_requests"

    if _looks_like_internal_resource_metric(measurement):
        raise ValueError("measurement must be user-facing, not internal resource metric")

    if not denominator:
        if sli_type in {SLIType.FRESHNESS.value, SLIType.COMPLETENESS.value}:
            denominator = "expected_records"
        elif sli_type == SLIType.CORRECTNESS.value:
            denominator = "processed_requests"
        else:
            denominator = "total_requests"

    dimension = kv.get("dimension") or "service,region"
    target_slo = kv.get("target_slo") or (
        "99.0% requests under threshold / rolling 30d"
        if sli_type == SLIType.LATENCY.value
        else "99.9% / rolling 30d"
    )
    error_budget = kv.get("error_budget") or (
        "1.0% / 30d"
        if sli_type == SLIType.LATENCY.value
        else "0.1% / 30d"
    )

    spec = SLISpec(
        capability=capability,
        user_journey=user_journey,
        sli_name=sli_name,
        sli_type=sli_type,
        measurement=measurement,
        denominator=denominator,
        dimension=dimension,
        target_slo=target_slo,
        error_budget=error_budget,
        severity=severity,
        owner=kv.get("owner") or owner,
    )

    notes: List[str] = []
    if "sli_type" not in kv:
        notes.append(f"inferred sli_type={spec.sli_type.value}")
    if "severity" not in kv:
        notes.append(f"inferred severity={spec.severity.value}")
    if "target_slo" not in kv:
        notes.append("defaulted target_slo to rolling 30d window")
    if "measurement" not in kv:
        notes.append("defaulted to user-facing request/data quality ratio")

    return ModelerResult(specs=[spec], notes=notes)

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

from __future__ import annotations

from dataclasses import asdict, dataclass
from enum import Enum
import re
from typing import Dict, Optional


_PERCENT_RE = re.compile(r"(\d+(?:\.\d+)?)\s*%")


def _extract_percent(value: str) -> Optional[float]:
    m = _PERCENT_RE.search(value)
    if not m:
        return None
    return float(m.group(1))


class SLIType(str, Enum):
    AVAILABILITY = "availability"
    LATENCY = "latency"
    CORRECTNESS = "correctness"
    FRESHNESS = "freshness"
    COMPLETENESS = "completeness"
    CONSISTENCY = "consistency"


class Severity(str, Enum):
    P0 = "P0"
    P1 = "P1"
    P2 = "P2"


@dataclass
class SLISpec:
    capability: str
    user_journey: str
    sli_name: str
    sli_type: SLIType
    measurement: str
    denominator: str
    dimension: str
    target_slo: str
    error_budget: str
    severity: Severity
    owner: str

    def __post_init__(self) -> None:
        for field_name in (
            "capability",
            "user_journey",
            "sli_name",
            "measurement",
            "denominator",
            "dimension",
            "target_slo",
            "error_budget",
            "owner",
        ):
            value = getattr(self, field_name)
            if not isinstance(value, str) or not value.strip():
                raise ValueError(f"{field_name} must be a non-empty string")
            setattr(self, field_name, value.strip())

        if isinstance(self.sli_type, str):
            self.sli_type = SLIType(self.sli_type.strip().lower())
        if isinstance(self.severity, str):
            self.severity = Severity(self.severity.strip().upper())

        target_percent = _extract_percent(self.target_slo)
        if target_percent is not None:
            if target_percent >= 100:
                raise ValueError("target_slo percent must be < 100%")
            if target_percent <= 0:
                raise ValueError("target_slo percent must be > 0%")

            budget_percent = _extract_percent(self.error_budget)
            if budget_percent is not None:
                expected_budget = 100 - target_percent
                if abs(expected_budget - budget_percent) > 0.1:
                    raise ValueError("error_budget percent must align with 1 - target_slo")

    def to_dict(self) -> Dict[str, str]:
        data = asdict(self)
        data["sli_type"] = self.sli_type.value
        data["severity"] = self.severity.value
        return data