Byted Sol Stability Architecture Path Extractor

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-architecture-path-extractor"
version = "0.1.0"
description = "Extract architecture topology, paths, dependency risks, and observability gaps"
readme = "SKILL.md"
requires-python = ">=3.10"
dependencies = ["pyyaml>=6.0"]

[project.scripts]
byted-sol-stablity-architecture-path-extractor = "architecture_path_extractor.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 __future__ import annotations

from typing import List

from .models import PathRecord, Signal


def classify_paths(signals: List[Signal]) -> List[PathRecord]:
    values = "\n".join([s.value + "\n" + s.evidence for s in signals]).lower()

    user_hops = ["login/auth", "create-space", "create-agent", "model-gateway", "tool-gateway", "return-result"]
    control_hops = ["config-change", "resource-dispatch", "permission-check", "rate-limit/quota", "audit"]
    data_hops = ["request-entry", "session-service", "orchestrator", "model-gateway", "tool-gateway", "memory/retrieval", "sandbox/browser", "persistence"]

    if "session" not in values:
        data_hops = ["request-entry", "orchestrator", "model-gateway", "persistence"]

    return [
        PathRecord(category="core_user_link", name="core_user_link", hops=user_hops, evidence=[s.source_file for s in signals[:5]]),
        PathRecord(category="control_plane_link", name="control_plane_link", hops=control_hops, evidence=[s.source_file for s in signals[:5]]),
        PathRecord(category="data_plane_link", name="data_plane_link", hops=data_hops, evidence=[s.source_file for s in signals[:8]]),
    ]

# 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 .pipeline import run_pipeline


def _parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="Extract architecture topology and paths")
    p.add_argument("--repo", required=True, help="Repository path")
    p.add_argument("--product-doc", action="append", default=[], help="Product document path")
    p.add_argument("--arch-diagram", action="append", default=[], help="Architecture diagram path")
    p.add_argument("--out-dir", default="output")
    p.add_argument("--focus-service")
    p.add_argument("--offline", action="store_true")
    return p


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

    model = run_pipeline(
        repo=str(repo_path),
        product_docs=args.product_doc,
        arch_diagrams=args.arch_diagram,
    )

    outdir = write_outputs(model=model, out_base_dir=args.out_dir, repo_slug=repo_path.name)
    report = {
        "output_dir": outdir,
        "service_nodes": len(model.service_graph.get("nodes", [])),
        "request_paths": len(model.request_paths),
        "async_paths": len(model.async_paths),
        "failure_points": len(model.failure_points),
        "observability_gaps": len(model.observability_hook_points),
    }
    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 Dict, List

from .models import TopologyModel


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(model: TopologyModel, out_base_dir: str, repo_slug: str) -> str:
    outdir = Path(out_base_dir) / _slug(repo_slug)
    outdir.mkdir(parents=True, exist_ok=True)

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

    core_lines = ["# Core Links", ""]
    core_lines.append("## Core User Links")
    for p in model.request_paths:
        core_lines.append(f"- {p['name']}: {' -> '.join(p['hops'])}")
    core_lines.append("")
    core_lines.append("## Control Plane Links")
    core_lines.append("- config-change -> resource-dispatch -> permission-check -> rate-limit/quota -> audit")
    core_lines.append("")
    core_lines.append("## Data Plane Links")
    for p in model.request_paths:
        if "data" in p["name"] or "core" in p["name"]:
            core_lines.append(f"- {p['name']}: {' -> '.join(p['hops'])}")
    (outdir / "core-links.md").write_text("\n".join(core_lines), encoding="utf-8")

    risk_lines = ["# Dependency Risk", ""]
    for f in model.failure_points:
        risk_lines.append(f"- {f['component']}: {f['risk']} ({f['impact']})")
    (outdir / "dependency-risk.md").write_text("\n".join(risk_lines), encoding="utf-8")

    gap_lines = ["# Observability Gaps", ""]
    for g in model.observability_hook_points:
        gap_lines.append(f"- {g['component']} [{g['gap_type']}]: {g['missing_signal']} -> {g['suggestion']}")
    (outdir / "observability-gaps.md").write_text("\n".join(gap_lines), encoding="utf-8")

    evidence: List[Dict[str, str]] = []
    for e in model.service_graph.get("edges", []):
        evidence.append({"kind": "edge", "source": e.get("evidence", "")})
    for p in model.request_paths + model.async_paths:
        for src in p.get("evidence", []):
            evidence.append({"kind": "path", "source": str(src)})
    (outdir / "evidence-index.json").write_text(
        json.dumps(evidence, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    return str(outdir)

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

from __future__ import annotations

from pathlib import Path
from typing import List
import re

from ..models import Signal


_ROUTE_RE = re.compile(r"\b(GET|POST|PUT|PATCH|DELETE)\b\s+(/[a-zA-Z0-9_\-/{}/:]*)")
_EXPRESS_RE = re.compile(r"\b(app|router)\.(get|post|put|patch|delete)\(\s*[\"']([^\"']+)")


def parse_api_signals(files: List[Path]) -> List[Signal]:
    signals: List[Signal] = []
    for f in files:
        if f.suffix.lower() not in {".py", ".ts", ".js", ".go", ".yaml", ".yml", ".json"}:
            continue
        try:
            text = f.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue

        for m in _ROUTE_RE.finditer(text):
            method = m.group(1)
            path = m.group(2)
            signals.append(Signal(kind="api_route", source_file=str(f), value=f"{method} {path}", evidence=m.group(0)))

        for m in _EXPRESS_RE.finditer(text):
            method = m.group(2).upper()
            path = m.group(3)
            signals.append(Signal(kind="api_route", source_file=str(f), value=f"{method} {path}", evidence=m.group(0)))

        if "openapi" in text.lower() and "paths:" in text.lower():
            signals.append(Signal(kind="openapi", source_file=str(f), value="openapi-spec", evidence="openapi + paths"))

    return signals

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

from __future__ import annotations

from pathlib import Path
from typing import List
import re

from ..models import Signal


_SERVICE_RE = re.compile(r"^\s{2,}([a-zA-Z0-9_-]+):\s*$")
_DEPEND_RE = re.compile(r"depends_on|redis|postgres|mysql|kafka|rabbitmq|mongo", re.IGNORECASE)


def parse_config_signals(files: List[Path]) -> List[Signal]:
    signals: List[Signal] = []
    for f in files:
        name = f.name.lower()
        if name not in {"docker-compose.yml", "docker-compose.yaml"} and not name.endswith((".yaml", ".yml", ".env")):
            continue
        try:
            text = f.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue

        for line in text.splitlines():
            m = _SERVICE_RE.search(line)
            if m:
                svc = m.group(1)
                signals.append(Signal(kind="service", source_file=str(f), value=svc, evidence=line.strip()))
            if _DEPEND_RE.search(line):
                signals.append(Signal(kind="dependency_hint", source_file=str(f), value=line.strip(), evidence=line.strip()))
    return signals

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

from __future__ import annotations

from pathlib import Path
from typing import List
import json
import re

from ..models import Signal


_MQ_RE = re.compile(r"kafka|rabbitmq|sqs|pubsub|queue|topic", re.IGNORECASE)
_DB_RE = re.compile(r"postgres|mysql|mongodb|redis|dynamodb|sqlite", re.IGNORECASE)


def parse_dependency_signals(files: List[Path]) -> List[Signal]:
    signals: List[Signal] = []
    for f in files:
        name = f.name.lower()
        if name == "package.json":
            try:
                payload = json.loads(f.read_text(encoding="utf-8", errors="ignore"))
            except Exception:
                continue
            deps = payload.get("dependencies", {})
            for dep in deps.keys():
                kind = "third_party"
                if _MQ_RE.search(dep):
                    kind = "mq"
                elif _DB_RE.search(dep):
                    kind = "db_or_cache"
                signals.append(Signal(kind=kind, source_file=str(f), value=dep, evidence=dep))
            continue

        if name in {"pnpm-workspace.yaml", "requirements.txt", "poetry.lock", "pom.xml", "go.mod"}:
            try:
                text = f.read_text(encoding="utf-8", errors="ignore")
            except OSError:
                continue
            for line in text.splitlines():
                raw = line.strip()
                if not raw:
                    continue
                if _MQ_RE.search(raw):
                    signals.append(Signal(kind="mq", source_file=str(f), value=raw, evidence=raw))
                elif _DB_RE.search(raw):
                    signals.append(Signal(kind="db_or_cache", source_file=str(f), value=raw, evidence=raw))
                elif "packages:" in raw or raw.startswith("-"):
                    signals.append(Signal(kind="workspace", source_file=str(f), value=raw, evidence=raw))
    return signals

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

from __future__ import annotations

from pathlib import Path
from typing import List
import re

from ..models import Signal


_KEYWORD_RE = re.compile(
    r"login|auth|agent|model|tool|session|orchestrator|gateway|memory|retrieval|sandbox|browser|audit|quota|rate limit|配置|权限|限流|审计",
    re.IGNORECASE,
)


def parse_doc_diagram_signals(paths: List[Path], kind: str) -> List[Signal]:
    signals: List[Signal] = []
    for p in paths:
        try:
            text = p.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        for line in text.splitlines():
            hit = _KEYWORD_RE.search(line)
            if hit:
                signals.append(Signal(kind=kind, source_file=str(p), value=hit.group(0).lower(), evidence=line.strip()))
    return signals

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

from __future__ import annotations

from pathlib import Path
from typing import List


def scan_repo_files(repo_path: Path) -> List[Path]:
    files: List[Path] = []
    for p in repo_path.rglob("*"):
        if p.is_dir():
            continue
        rel = p.relative_to(repo_path)
        if any(part.startswith(".") for part in rel.parts):
            continue
        if "node_modules" in rel.parts:
            continue
        files.append(p)
    return files

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

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import List


@dataclass
class InputBundle:
    repo_path: Path
    product_docs: List[Path] = field(default_factory=list)
    arch_diagrams: List[Path] = field(default_factory=list)


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

    docs = [Path(x) for x in product_docs if Path(x).exists() and Path(x).is_file()]
    diagrams = [Path(x) for x in arch_diagrams if Path(x).exists() and Path(x).is_file()]
    return InputBundle(repo_path=repo_path, product_docs=docs, arch_diagrams=diagrams)

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

from __future__ import annotations

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


@dataclass
class Signal:
    kind: str
    source_file: str
    value: str
    evidence: str = ""


@dataclass
class GraphNode:
    node_id: str
    node_type: str
    name: str
    attributes: Dict[str, str] = field(default_factory=dict)


@dataclass
class GraphEdge:
    source: str
    target: str
    edge_type: str
    evidence: str


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


@dataclass
class FailurePoint:
    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 TopologyModel:
    service_graph: Dict[str, List[Dict[str, str]]]
    request_paths: List[Dict[str, object]]
    async_paths: List[Dict[str, object]]
    dependency_graph: Dict[str, List[Dict[str, str]]]
    failure_points: List[Dict[str, object]]
    observability_hook_points: List[Dict[str, object]]
    metadata: Dict[str, str]

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

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

from __future__ import annotations

from typing import List

from .models import FailurePoint, ObservabilityGap


def find_observability_gaps(failures: List[FailurePoint], request_paths: List[dict], async_paths: List[dict]) -> List[ObservabilityGap]:
    gaps: List[ObservabilityGap] = []

    for f in failures:
        gaps.append(
            ObservabilityGap(
                component=f.component,
                gap_type="metric",
                missing_signal=f"{f.component}_availability or saturation",
                suggestion=f"add RED + saturation metrics for {f.component}",
                evidence=f.evidence,
            )
        )

    if request_paths:
        gaps.append(
            ObservabilityGap(
                component="request_path",
                gap_type="trace",
                missing_signal="end-to-end trace span across core user path",
                suggestion="add trace span propagation at gateway/orchestrator/model/tool boundaries",
                evidence=request_paths[0].get("evidence", [])[:3],
            )
        )

    if async_paths:
        gaps.append(
            ObservabilityGap(
                component="async_path",
                gap_type="log",
                missing_signal="producer-consumer correlation id logging",
                suggestion="add structured logs with correlation_id for async producer/consumer",
                evidence=async_paths[0].get("evidence", [])[:3],
            )
        )

    gaps.append(
        ObservabilityGap(
            component="control_plane",
            gap_type="audit",
            missing_signal="config change and permission-check audit trail",
            suggestion="add immutable audit events for config changes, quota and permission decisions",
            evidence=[],
        )
    )

    return gaps

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

from __future__ import annotations

from typing import Dict, List

from .models import GraphEdge, GraphNode, PathRecord, Signal


def build_service_graph(signals: List[Signal]) -> tuple[List[GraphNode], List[GraphEdge]]:
    services = sorted({s.value for s in signals if s.kind == "service"})
    if not services:
        services = ["request-entry", "session-service", "orchestrator", "model-gateway", "tool-gateway", "persistence"]

    nodes: List[GraphNode] = [GraphNode(node_id=f"svc:{x}", node_type="service", name=x) for x in services]
    edges: List[GraphEdge] = []
    for i in range(len(services) - 1):
        edges.append(
            GraphEdge(
                source=f"svc:{services[i]}",
                target=f"svc:{services[i + 1]}",
                edge_type="call",
                evidence="service adjacency from config/order",
            )
        )
    return nodes, edges


def build_dependency_graph(signals: List[Signal]) -> Dict[str, List[Dict[str, str]]]:
    out: Dict[str, List[Dict[str, str]]] = {
        "db": [],
        "cache": [],
        "mq": [],
        "third_party": [],
    }
    for s in signals:
        v = s.value.lower()
        item = {"name": s.value, "source": s.source_file}
        if s.kind == "mq" or any(k in v for k in ["kafka", "rabbitmq", "queue", "topic", "sqs", "pubsub"]):
            out["mq"].append(item)
        elif any(k in v for k in ["postgres", "mysql", "mongo", "dynamodb", "sqlite"]):
            out["db"].append(item)
        elif any(k in v for k in ["redis", "memcached", "cache"]):
            out["cache"].append(item)
        elif s.kind in {"third_party", "db_or_cache", "dependency_hint"}:
            out["third_party"].append(item)

    for key in out:
        seen = set()
        deduped = []
        for item in out[key]:
            name = item["name"]
            if name in seen:
                continue
            seen.add(name)
            deduped.append(item)
        out[key] = deduped
    return out


def build_request_and_async_paths(path_records: List[PathRecord], signals: List[Signal]) -> tuple[List[Dict[str, object]], List[Dict[str, object]]]:
    request_paths: List[Dict[str, object]] = []
    async_paths: List[Dict[str, object]] = []
    for p in path_records:
        if p.category in {"core_user_link", "data_plane_link"}:
            request_paths.append({"name": p.name, "hops": p.hops, "evidence": p.evidence})

    mq_hits = [s for s in signals if s.kind == "mq" or "queue" in s.value.lower() or "topic" in s.value.lower()]
    if mq_hits:
        async_paths.append(
            {
                "name": "async_dependency_path",
                "hops": ["producer", "mq/topic", "consumer", "persistence"],
                "evidence": [x.source_file for x in mq_hits[:8]],
            }
        )
    return request_paths, async_paths

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

from __future__ import annotations

from pathlib import Path
from typing import List

from .classifier import classify_paths
from .extractors.api_parser import parse_api_signals
from .extractors.config_parser import parse_config_signals
from .extractors.dependency_parser import parse_dependency_signals
from .extractors.doc_diagram_parser import parse_doc_diagram_signals
from .extractors.repo_scanner import scan_repo_files
from .inputs import build_input_bundle
from .models import TopologyModel
from .observability_gap import find_observability_gaps
from .path_builder import build_dependency_graph, build_request_and_async_paths, build_service_graph
from .risk_analyzer import find_failure_points


def run_pipeline(repo: str, product_docs: List[str], arch_diagrams: List[str]) -> TopologyModel:
    bundle = build_input_bundle(repo=repo, product_docs=product_docs, arch_diagrams=arch_diagrams)
    files = scan_repo_files(bundle.repo_path)

    config_signals = parse_config_signals(files)
    api_signals = parse_api_signals(files)
    dep_signals = parse_dependency_signals(files)
    doc_signals = parse_doc_diagram_signals(bundle.product_docs, kind="product_doc")
    diagram_signals = parse_doc_diagram_signals(bundle.arch_diagrams, kind="arch_diagram")

    signals = config_signals + api_signals + dep_signals + doc_signals + diagram_signals

    nodes, edges = build_service_graph(signals)
    dependency_graph = build_dependency_graph(signals)
    path_records = classify_paths(signals)
    request_paths, async_paths = build_request_and_async_paths(path_records, signals)
    failures = find_failure_points(dependency_graph, request_paths)
    gaps = find_observability_gaps(failures, request_paths, async_paths)

    model = TopologyModel(
        service_graph={
            "nodes": [
                {"id": n.node_id, "type": n.node_type, "name": n.name, "attributes": n.attributes}
                for n in nodes
            ],
            "edges": [
                {"source": e.source, "target": e.target, "edge_type": e.edge_type, "evidence": e.evidence}
                for e in edges
            ],
        },
        request_paths=request_paths,
        async_paths=async_paths,
        dependency_graph=dependency_graph,
        failure_points=[
            {
                "component": f.component,
                "risk": f.risk,
                "impact": f.impact,
                "evidence": f.evidence,
            }
            for f in failures
        ],
        observability_hook_points=[
            {
                "component": g.component,
                "gap_type": g.gap_type,
                "missing_signal": g.missing_signal,
                "suggestion": g.suggestion,
                "evidence": g.evidence,
            }
            for g in gaps
        ],
        metadata={
            "repo": str(bundle.repo_path),
            "signals": str(len(signals)),
            "docs": str(len(bundle.product_docs)),
            "diagrams": str(len(bundle.arch_diagrams)),
        },
    )
    return model

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

from __future__ import annotations

from typing import Dict, List

from .models import FailurePoint


def find_failure_points(dependency_graph: Dict[str, List[Dict[str, str]]], request_paths: List[Dict[str, object]]) -> List[FailurePoint]:
    failures: List[FailurePoint] = []

    if dependency_graph.get("db"):
        failures.append(
            FailurePoint(
                "database",
                "single critical datastore",
                "core read/write path degraded or unavailable",
                evidence=[x.get("source", "") for x in dependency_graph["db"][:3]],
            )
        )

    if dependency_graph.get("mq"):
        failures.append(
            FailurePoint(
                "message-queue",
                "async backlog or broker outage",
                "delayed task execution and eventual user-visible timeout",
                evidence=[x.get("source", "") for x in dependency_graph["mq"][:3]],
            )
        )

    if request_paths:
        failures.append(
            FailurePoint(
                "orchestrator",
                "central orchestration bottleneck",
                "major user journey interruption",
                evidence=request_paths[0].get("evidence", [])[:3],
            )
        )

    return failures