from __future__ import annotations import json import shutil import subprocess from pathlib import Path from techviz.layout import build_layout from techviz.renderers.svg import render_svg from techviz.spec import VizSpec from techviz.validate import has_errors, validate_spec ROOT = Path(__file__).resolve().parent OUT = ROOT / "runtime-profiles" E = [{"start_line": 1, "end_line": 1}] SOURCE = { "document": "examples/runtime-profiles/fixture.md", "document_sha256": "0" * 64, "anchor": {"kind": "line", "value": 1, "line": 1}, } def node(node_id: str, label: str, kind: str, role: str, **extra: object) -> dict[str, object]: payload: dict[str, object] = { "id": node_id, "label": label, "kind": kind, "role": role, "evidence": E, "assumption": False, } payload.update(extra) return payload def edge(edge_id: str, source: str, target: str, label: str, kind: str = "request", **extra: object) -> dict[str, object]: payload: dict[str, object] = { "id": edge_id, "from": source, "to": target, "label": label, "kind": kind, "evidence": E, "assumption": False, } payload.update(extra) return payload def spec(profile: str, ref: str, diagram_type: str, direction: str, nodes: list[dict[str, object]], edges: list[dict[str, object]], *, groups: list[dict[str, object]] | None = None, focus: str | None = None) -> dict[str, object]: composition: dict[str, object] = { "profile": profile, "diagram_only": True, "reference_ids": [ref], "rationale": f"Runtime regression fixture for {profile}.", } if focus: composition["focus_node"] = focus return { "version": "1.1", "id": profile, "title": f"{profile} runtime fixture", "question": f"Does the {profile} logical composition render correctly?", "type": diagram_type, "direction": direction, "audience": ["renderer maintainer"], "summary": f"Executable {profile} fixture.", "alt": f"Executable technical diagram fixture for the {profile} composition profile.", "long_description": f"This fixture validates the node roles, edges, and layout rules of the {profile} profile.", "source_context": SOURCE, "composition": composition, "groups": groups or [], "nodes": nodes, "edges": edges, "legend": [], "metadata": {"fixture": True}, } FIXTURES: dict[str, dict[str, object]] = { "01-component-flow": spec( "component-flow", "payment-event-flow", "data-flow", "LR", [ node("client", "클라이언트", "actor", "source", shape="actor"), node("gateway", "인증 게이트웨이", "security", "service"), node("api", "체크아웃 API", "service", "service", emphasis="primary"), node("db", "주문 저장소", "database", "store", shape="database"), node("provider", "결제 제공자", "service", "service"), node("bus", "이벤트 버스", "queue", "sink", shape="queue"), ], [ edge("request", "client", "gateway", "HTTPS 결제 요청"), edge("validated", "gateway", "api", "검증된 요청"), edge("pending", "api", "db", "PENDING 기록", "data"), edge("authorize", "api", "provider", "승인 요청"), edge("approved", "provider", "api", "승인 응답", "response", style="dashed"), edge("paid", "api", "db", "PAID 갱신", "data"), edge("event", "api", "bus", "payment.approved", "publish", style="dashed", emphasis="primary"), ], focus="api", ), "02-orchestrator-workers": spec( "orchestrator-workers", "mission-workers", "architecture", "TB", [ node("main", "MAIN SESSION", "service", "orchestrator", shape="controller", emphasis="primary", details=["context + logic"]), node("agents", "SUBAGENTS × N", "service", "worker", details=["client RPS · latency", "server RPS · latency", "JVM heap · GC"]), node("stream", "BACKGROUND BASH", "process", "subprocess", details=["tail -f ..."]), node("poll", "POLLING BASH", "process", "subprocess", details=["until ; sleep N"]), node("monitor", "MONITOR", "service", "monitor", shape="controller", details=["stdout 수신", "이벤트 알림"]), ], [ edge("dispatch", "main", "agents", "dispatch"), edge("spawn-stream", "main", "stream", "spawn ×2"), edge("spawn-poll", "main", "poll", "spawn"), edge("subscribe", "main", "monitor", "subscribe"), edge("stdout-stream", "stream", "monitor", "stdout"), edge("stdout-poll", "poll", "monitor", "stdout"), edge("notify", "monitor", "main", "notification", "notification", style="dashed"), ], focus="main", ), "03-query-fanout": spec( "query-fanout", "metrics-query-fanout", "data-flow", "LR", [ node("user", "사용자", "actor", "actor", shape="actor"), node("query", "rate(http_requests_total[5m])", "request", "query", details=["time range: 15:00–15:10"]), node("parser", "query parser", "component", "parser", shape="port"), node("selector", "vmselect", "service", "router", emphasis="primary"), node("a", "vmstorage A", "database", "shard", shape="database", details=["IndexDB → TSID → Data"]), node("b", "vmstorage B", "database", "shard", shape="database", details=["IndexDB → TSID → Data"]), node("c", "vmstorage C", "database", "shard", shape="database", details=["IndexDB → TSID → Data"]), ], [ edge("submit", "user", "query", "PromQL"), edge("parse", "query", "parser", "parse"), edge("select", "parser", "selector", "filters + time range"), edge("to-a", "selector", "a", "range query"), edge("to-b", "selector", "b", "range query"), edge("to-c", "selector", "c", "range query"), ], focus="selector", ), "04-timeline": spec( "timeline", "retention-cycle", "concept", "LR", [ node("t0", "T = 0", "milestone", "milestone", position=1, details=["1970-01-01", "Unix Epoch"]), node("d372", "+372d", "milestone", "milestone", position=2, details=["1971-01-08"]), node("d744", "+744d", "milestone", "milestone", position=3, details=["1972-01-15"]), node("n55", "n=55", "milestone", "milestone", position=4, emphasis="warning", details=["2026-01-07", "직전 로테이션"]), node("today", "현재", "milestone", "milestone", position=5, emphasis="primary", details=["오늘"]), node("n56", "n=56", "milestone", "milestone", position=6, details=["2027-01-14", "다음 로테이션"]), ], [], focus="today", ), "05-reconciliation-loop": spec( "reconciliation-loop", "declarative-vm", "architecture", "LR", [ node("desired", "VM Custom Resource", "resource", "desired-state", shape="document", details=["vCPU: 2", "memory: 8GB", "process: dbaas-agent"]), node("controller", "VM Operator", "controller", "controller", shape="controller", emphasis="primary"), node("actual", "VM", "runtime", "actual-state", details=["dbaas-agent"]), node("status", "status", "state", "status", shape="port", details=["dbaas-agent: ok"]), ], [ edge("watch", "desired", "controller", "watch"), edge("create", "controller", "actual", "create VM", emphasis="primary"), edge("status-read", "actual", "status", "report status", "response", style="dashed"), edge("status-write", "status", "desired", "update status", "response", style="dashed"), ], focus="controller", ), "06-resource-controller": spec( "resource-controller", "dbaas-controller", "architecture", "LR", [ node("user", "User", "actor", "actor", shape="actor"), node("db-service", "DB Service", "resource", "resource-spec", shape="document", details=["vmCount: 3", "vCPU: 2", "memory: 8GB", "maxclient: 3000"]), node("manager", "DBaaS Manager", "controller", "controller", shape="controller", emphasis="primary"), node("vm-cr", "VM Custom Resource ×3", "resource", "custom-resource", shape="document", details=["2 vCPU", "8GB RAM"]), node("db-cr", "DB Instance CR ×3", "resource", "custom-resource", shape="document", details=["maxclient: 3000", "timeout: 5s"]), node("vm", "VM + dbaas-agent", "runtime", "runtime-resource", details=["create VM"]), node("db", "DB", "database", "runtime-resource", shape="database", details=["DB command"]), ], [ edge("create-spec", "user", "db-service", "create"), edge("watch-spec", "db-service", "manager", "watch"), edge("create-vm-cr", "manager", "vm-cr", "create ×3"), edge("create-db-cr", "manager", "db-cr", "create ×3"), edge("materialize-vm", "vm-cr", "vm", "reconcile"), edge("materialize-db", "db-cr", "db", "reconcile"), edge("agent-command", "vm", "db", "DB command"), ], focus="manager", ), "07-two-zone-pipeline": spec( "two-zone-pipeline", "localization-pipeline", "data-flow", "LR", [ node("mobile", "배민 앱", "actor", "source", group="request-zone", shape="actor"), node("list", "목록/상세 BFF", "service", "service", group="request-zone", details=["lang param"]), node("cart", "장바구니/주문 BFF", "service", "service", group="request-zone", details=["lang param"]), node("fdh", "FDH", "service", "service", group="request-zone", emphasis="primary"), node("source-data", "가게/메뉴", "database", "store", group="translation-zone", shape="database"), node("worker", "FDH Worker", "service", "service", group="translation-zone"), node("queue", "Queue", "queue", "queue", group="translation-zone", shape="queue"), node("translator", "LLM Translator", "service", "service", group="translation-zone"), ], [ edge("app-list", "mobile", "list", "목록/상세 요청"), edge("app-cart", "mobile", "cart", "주문 요청"), edge("list-fdh", "list", "fdh", "다국어 조회"), edge("cart-fdh", "cart", "fdh", "다국어 조회"), edge("data-worker", "source-data", "worker", "이벤트"), edge("worker-queue", "worker", "queue", "적재 작업"), edge("queue-translator", "queue", "translator", "번역 요청"), edge("translator-worker", "translator", "worker", "번역 결과", "response", style="dashed"), edge("worker-fdh", "worker", "fdh", "다국어 적재", "data"), ], groups=[ {"id": "request-zone", "label": "사용자 요청 및 서비스 레이어", "kind": "system", "role": "zone", "evidence": E, "assumption": False}, {"id": "translation-zone", "label": "데이터 번역 및 적재 레이어", "kind": "system", "role": "zone", "evidence": E, "assumption": False}, ], focus="fdh", ), "08-sequence": spec( "sequence", "payment-approval-sequence", "sequence", "LR", [ node("client", "Client", "participant", "participant"), node("api", "Checkout API", "participant", "participant", emphasis="primary"), node("provider", "Payment Provider", "participant", "participant"), node("db", "Orders DB", "participant", "participant"), node("bus", "Event Bus", "participant", "participant"), ], [ edge("m1", "client", "api", "결제 요청", order=1), edge("m2", "api", "db", "PENDING 기록", "data", order=2), edge("m3", "api", "provider", "승인 요청", order=3), edge("m4", "provider", "api", "승인 응답", "response", order=4, style="dashed"), edge("m5", "api", "db", "PAID 갱신", "data", order=5), edge("m6", "api", "bus", "payment.approved", "publish", order=6, style="dashed", emphasis="primary"), ], focus="api", ), "09-ports-adapters": spec( "ports-adapters", "order-ports-adapters", "architecture", "LR", [ node("web", "Web Adapter", "adapter", "inbound-adapter", details=["REST · inbound"]), node("batch", "Test / Batch", "adapter", "inbound-adapter", details=["inbound"]), node("in-port", "Input Port", "interface", "port", shape="port"), node("core", "Application Core", "domain", "core", shape="hexagon", emphasis="primary", details=["domain + use cases"]), node("out-port", "Output Port", "interface", "port", shape="port"), node("persistence", "Persistence Adapter", "adapter", "outbound-adapter", details=["JPA · outbound"]), node("db", "DB", "database", "external-system", shape="database"), ], [ edge("web-port", "web", "in-port", "call"), edge("batch-port", "batch", "in-port", "call"), edge("port-core", "in-port", "core", "invoke use case", emphasis="primary"), edge("core-out", "core", "out-port", "requires"), edge("adapter-out", "persistence", "out-port", "implements", "dependency"), edge("adapter-db", "persistence", "db", "JDBC"), ], focus="core", ), "10-comparison": spec( "comparison", "contract-comparison", "concept", "LR", [ node("transaction", "TransactionPort", "interface", "contract", details=["세션 모델: Spring 구현", "책임: commit / rollback", "현재: 독립 계약"], emphasis="primary"), node("lock", "DistributedLockPort", "interface", "contract", details=["세션 모델: 호출자 0", "책임: acquire / release", "현재: 독립 계약"], emphasis="primary"), node("future", "Future lifecycle", "concept", "option", details=["acquire → commit → release", "통합 시 순서 명시 필요"], emphasis="warning"), ], [], ), } def main() -> int: OUT.mkdir(parents=True, exist_ok=True) inkscape = shutil.which("inkscape") for directory, raw in FIXTURES.items(): target = OUT / directory target.mkdir(parents=True, exist_ok=True) spec_obj = VizSpec.from_dict(raw) issues = validate_spec(spec_obj) if has_errors(issues): raise RuntimeError(f"{directory}: {issues}") (target / "spec.json").write_text(json.dumps(raw, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") svg_path = target / f"{spec_obj.id}.svg" svg_path.write_text(render_svg(spec_obj, build_layout(spec_obj)), encoding="utf-8") if inkscape: subprocess.run( [inkscape, str(svg_path), "--export-type=png", f"--export-filename={target / (spec_obj.id + '.preview.png')}"], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) print(f"WROTE {svg_path}") return 0 if __name__ == "__main__": raise SystemExit(main())