#!/usr/bin/env python3 """Fail-closed comparison of three equivalent Runtime Worker observations.""" from __future__ import annotations import argparse import csv import hashlib import json import math import re import sys from datetime import datetime from pathlib import Path from typing import Any, Iterable, Sequence sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.serviceification_d3_engine import ( # noqa: E402 LEGACY_WORKER_STEMS as WORKER_STEMS, READABLE_WORKER_STEM_SETS as SUPPORTED_WORKER_STEM_SETS, ) SCENARIO_CASES = { "quiet": "queues-empty", "mixed-workload": "all-supported-families", } MAX_RELATIVE_SPREAD = 0.10 SUPPORTED_TARGET_ENVS = {"test2", "test3"} WORKLOAD_OWNERS = { "mixed-workload": { "name": "Env Inspection", "path": ".github/workflows/staging-inspection.yml", "event": "workflow_dispatch", }, } IMAGE_ROLES = { "runtime_api", "runtime_worker", "integration_service", "sfcrm_app_service", "catalog_reconcile", "frontend", } GAUGE_COSTS = ( "rss_peak_bytes", "heap_peak_bytes", "goroutines_peak", "primary_pool_open_peak", "primary_pool_in_use_peak", "queue_observation_pool_open_peak", "rejected_save_pool_open_peak", ) EXECUTION_COSTS = ( "poll_checks_delta", "claim_attempts_delta", "claimed_delta", "processed_delta", "attempts_delta", ) QUIET_ZERO_GAUGES = ( "queue_depth_peak", "poison_depth_peak", "in_flight_peak", ) QUIET_ZERO_COUNTERS = ( "claimed_delta", "processed_delta", "failures_delta", "scheduled_accepted_delta", "scheduled_execution_started_delta", "scheduled_execution_completed_delta", "scheduled_execution_failed_delta", ) class BaselineError(RuntimeError): pass def load_json(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text()) except (OSError, json.JSONDecodeError) as exc: raise BaselineError(f"cannot read {path}: {exc}") from exc if not isinstance(value, dict): raise BaselineError(f"{path} is not a JSON object") return value def validate_scenario_case(scenario: Any, baseline_case: Any) -> None: expected = SCENARIO_CASES.get(scenario) if expected is None or baseline_case != expected: raise BaselineError("baseline scenario/case is unsupported") def validate_full_inspection_report(report: dict[str, Any]) -> dict[str, Any]: environment = report.get("environment") planned = ( environment.get("caseInventory") if isinstance(environment, dict) else None ) roles = report.get("roles") summary = report.get("summary") optional_case_keys = ( planned.get("optionalCaseKeys") if isinstance(planned, dict) else None ) if ( not isinstance(roles, list) or not isinstance(summary, dict) or not isinstance(planned, dict) or planned.get("schema") != "nexuscrm.inspection-case-inventory/v1" or not isinstance(planned.get("caseCount"), int) or isinstance(planned.get("caseCount"), bool) or planned["caseCount"] <= 0 or not isinstance(planned.get("digest"), str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", planned["digest"]) or not isinstance(optional_case_keys, list) or any(not isinstance(key, str) or not key for key in optional_case_keys) or optional_case_keys != sorted(set(optional_case_keys)) ): raise BaselineError("inspection report has no case inventory") observed_roles: set[str] = set() observed_modules: set[str] = set() observed_cases: dict[str, str] = {} for role in roles: role_id = role.get("id") if isinstance(role, dict) else None modules = role.get("modules") if isinstance(role, dict) else None if ( not isinstance(role_id, str) or not role_id or role_id in observed_roles or not isinstance(modules, list) ): raise BaselineError("inspection report has a malformed role") observed_roles.add(role_id) for module in modules: module_id = module.get("id") if isinstance(module, dict) else None functions = module.get("functions") if isinstance(module, dict) else None module_key = f"{role_id}/{module_id}" if ( not isinstance(module_id, str) or not module_id or module_key in observed_modules or not isinstance(functions, list) ): raise BaselineError("inspection report has a malformed module") observed_modules.add(module_key) for function in functions: function_id = ( function.get("id") if isinstance(function, dict) else None ) status = ( function.get("status") if isinstance(function, dict) else None ) case_key = f"{module_key}/{function_id}" if ( not isinstance(function_id, str) or not function_id or status not in {"passed", "failed", "skipped"} or case_key in observed_cases ): raise BaselineError("inspection report has a malformed case") observed_cases[case_key] = status if any(key not in observed_cases for key in optional_case_keys): raise BaselineError("inspection report has a malformed optional case inventory") optional_cases = set(optional_case_keys) canonical = [ {"key": key, "optional": key in optional_cases} for key in sorted(observed_cases) ] observed_digest = "sha256:" + hashlib.sha256( (json.dumps(canonical, separators=(",", ":")) + "\n").encode() ).hexdigest() if ( planned["caseCount"] != len(observed_cases) or planned["digest"] != observed_digest ): raise BaselineError("inspection report does not cover the full case inventory") status_counts = { status: sum(value == status for value in observed_cases.values()) for status in ("passed", "failed", "skipped") } if ( summary.get("roleTotal") != len(observed_roles) or summary.get("moduleTotal") != len(observed_modules) or summary.get("functionTotal") != len(observed_cases) or any(summary.get(status) != count for status, count in status_counts.items()) ): raise BaselineError("inspection report summary does not match its case inventory") optional_skips = sorted( key for key, status in observed_cases.items() if status == "skipped" ) if any(key not in optional_cases for key in optional_skips): raise BaselineError("inspection report skipped a required case") if status_counts["failed"] != 0 or status_counts["passed"] <= 0: raise BaselineError("inspection report did not complete successfully") return { "caseInventoryDigest": planned["digest"], "caseCount": planned["caseCount"], "passedCaseCount": status_counts["passed"], "optionalCaseKeys": optional_case_keys, "optionalSkippedCaseKeys": optional_skips, } def payload_root(root: Path) -> Path: for candidate in (root, root / "logs"): if ( (candidate / "runtime-worker-baseline-identity.json").is_file() and (candidate / "runtime-worker-baseline-context.json").is_file() and (candidate / "observation" / "summary.json").is_file() ): return candidate raise BaselineError(f"{root} has no complete Runtime Worker baseline artifact") def finite_number(value: Any, owner: str) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise BaselineError(f"{owner} is not numeric") result = float(value) if not math.isfinite(result) or result < 0: raise BaselineError(f"{owner} is not one finite non-negative value") return result def required_number(values: Any, name: str, owner: str) -> float: if not isinstance(values, dict) or name not in values: raise BaselineError(f"{owner}.{name} is missing") return finite_number(values[name], f"{owner}.{name}") def tsv_number(value: Any, owner: str) -> float: if not isinstance(value, str) or not value.strip(): raise BaselineError(f"{owner} is missing") try: result = float(value) except ValueError as exc: raise BaselineError(f"{owner} is not numeric") from exc if not math.isfinite(result) or result < 0: raise BaselineError(f"{owner} is not one finite non-negative value") return result def window_seconds(summary: dict[str, Any]) -> float: try: first = datetime.fromisoformat(str(summary["first_at"]).replace("Z", "+00:00")) last = datetime.fromisoformat(str(summary["last_at"]).replace("Z", "+00:00")) except (KeyError, ValueError) as exc: raise BaselineError("observation has an invalid time window") from exc seconds = (last - first).total_seconds() if seconds <= 0: raise BaselineError("observation time window is not positive") return seconds def parse_instant(value: Any, owner: str) -> datetime: if not isinstance(value, str): raise BaselineError(f"{owner} is missing") try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise BaselineError(f"{owner} is not one ISO-8601 instant") from exc def validate_workload_run( raw: dict[str, Any], *, scenario: str, workload_run_id: str, observer_revision: str, observed_start: datetime, observed_end: datetime, report: dict[str, Any], target_env: str = "test2", ) -> dict[str, Any]: if target_env not in SUPPORTED_TARGET_ENVS: raise BaselineError("baseline workload target is unsupported") if not isinstance(raw, dict) or not isinstance(report, dict): raise BaselineError("baseline workload evidence is not a JSON object") expected = WORKLOAD_OWNERS.get(scenario) if expected is None: raise BaselineError(f"{scenario} has no trusted workload owner") status = raw.get("status") workload_start = parse_instant(raw.get("run_started_at"), "workload start") if status != "completed" or raw.get("conclusion") != "success": raise BaselineError("baseline workload must complete successfully") workload_end = parse_instant(raw.get("updated_at"), "workload end") workflow_id = raw.get("workflow_id") if ( raw.get("id") != int(workload_run_id) or raw.get("head_sha") != observer_revision or any(raw.get(field) != value for field, value in expected.items()) or not isinstance(workflow_id, int) or isinstance(workflow_id, bool) or workflow_id <= 0 or workload_start > observed_end or workload_end < observed_start ): raise BaselineError( "baseline workload owner, revision, or time window does not match" ) environment = report.get("environment") selection = environment.get("selection") if isinstance(environment, dict) else None if ( not isinstance(environment, dict) or environment.get("targetEnvironment") != target_env or environment.get("targetSource") != target_env or environment.get("inspectionCheckoutSha") != observer_revision or selection != {"role": "", "functions": [], "cases": []} ): raise BaselineError( "baseline workload report is not one successful full selected-Test inspection" ) case_evidence = validate_full_inspection_report(report) return { "runId": workload_run_id, "workflowId": workflow_id, "name": expected["name"], "path": expected["path"], "event": expected["event"], "headRevision": observer_revision, "inspectionRunId": report.get("runId"), "inspectionReportDigest": "sha256:" + hashlib.sha256( (json.dumps(report, sort_keys=True, separators=(",", ":")) + "\n").encode() ).hexdigest(), "selection": selection, "targetEnvironment": target_env, "startedAt": raw["run_started_at"], "endedAt": raw.get("updated_at"), "statusAtCollection": status, "conclusion": raw.get("conclusion"), **case_evidence, } def validate_context_workload( workload: Any, *, scenario: str, workload_run_id: str, observer_revision: str, summary: dict[str, Any], target_env: str = "test2", ) -> None: if target_env not in SUPPORTED_TARGET_ENVS: raise BaselineError("baseline workload target is unsupported") if summary_target_env(summary) != target_env: raise BaselineError("baseline workload and observation target different Test slots") expected = WORKLOAD_OWNERS.get(scenario) if not isinstance(workload, dict) or expected is None: raise BaselineError("baseline workload binding is incomplete") status = workload.get("statusAtCollection") started = parse_instant(workload.get("startedAt"), "bound workload start") observed_start = parse_instant(summary.get("first_at"), "observation start") observed_end = parse_instant(summary.get("last_at"), "observation end") ended = parse_instant(workload.get("endedAt"), "bound workload end") optional_cases = workload.get("optionalCaseKeys") optional_skips = workload.get("optionalSkippedCaseKeys") if ( workload.get("runId") != workload_run_id or workload.get("headRevision") != observer_revision or any(workload.get(field) != value for field, value in expected.items()) or not isinstance(workload.get("workflowId"), int) or isinstance(workload.get("workflowId"), bool) or workload["workflowId"] <= 0 or status != "completed" or workload.get("conclusion") != "success" or workload.get("targetEnvironment") != target_env or workload.get("selection") != {"role": "", "functions": [], "cases": []} or not isinstance(workload.get("caseInventoryDigest"), str) or not re.fullmatch( r"sha256:[0-9a-f]{64}", workload["caseInventoryDigest"] ) or not isinstance(workload.get("caseCount"), int) or isinstance(workload.get("caseCount"), bool) or workload["caseCount"] <= 0 or not isinstance(workload.get("passedCaseCount"), int) or isinstance(workload.get("passedCaseCount"), bool) or workload["passedCaseCount"] <= 0 or not isinstance(optional_cases, list) or any(not isinstance(key, str) or not key for key in optional_cases) or optional_cases != sorted(set(optional_cases)) or not isinstance(optional_skips, list) or any(not isinstance(key, str) or not key for key in optional_skips) or optional_skips != sorted(set(optional_skips)) or any(key not in optional_cases for key in optional_skips) or workload.get("passedCaseCount") + len(optional_skips) != workload.get("caseCount") or not isinstance(workload.get("inspectionRunId"), str) or not workload["inspectionRunId"] or not isinstance(workload.get("inspectionReportDigest"), str) or not re.fullmatch( r"sha256:[0-9a-f]{64}", workload["inspectionReportDigest"] ) or started > observed_end or ended < observed_start ): raise BaselineError("baseline workload binding is incomplete") def statement_costs(path: Path) -> dict[str, float]: try: handle = path.open(newline="") except OSError as exc: raise BaselineError(f"cannot read {path}: {exc}") from exc with handle: reader = csv.DictReader(handle, delimiter="\t") required = {"executions", "avg_request_unit_read", "evidence_owner"} if reader.fieldnames is None or not required <= set(reader.fieldnames): raise BaselineError("statement digest evidence is missing cost columns") rows = list(reader) if not rows: raise BaselineError("statement digest evidence contains no statements") result = { "sql.digest_detail_executions": 0.0, "sql.digest_detail_estimated_read_ru": 0.0, } for index, row in enumerate(rows, start=2): executions = tsv_number(row.get("executions"), f"statement row {index} executions") average_ru = tsv_number( row.get("avg_request_unit_read"), f"statement row {index} average read RU", ) result["sql.digest_detail_executions"] += executions result["sql.digest_detail_estimated_read_ru"] += executions * average_ru return result def summary_target_env(summary: dict[str, Any]) -> str: schema = summary.get("schema") inferred = next( ( target for target in sorted(SUPPORTED_TARGET_ENVS) if schema == f"nexuscrm.{target}-runtime-observation/v2" ), "", ) explicit = summary.get("target_env", inferred) if inferred not in SUPPORTED_TARGET_ENVS or explicit != inferred: raise BaselineError("observation summary Test authority is unsupported") return inferred def validate_summary(summary: dict[str, Any]) -> tuple[str, ...]: summary_target_env(summary) workers = summary.get("workers") worker_stems = tuple(sorted(workers)) if isinstance(workers, dict) else () if ( summary.get("accepted") is not True or summary.get("cancelled") is not False or not worker_stems or frozenset(worker_stems) not in SUPPORTED_WORKER_STEM_SETS or summary.get("expected_worker_count") != len(worker_stems) or summary.get("expected_container_count") != len(worker_stems) + 3 or summary.get("runtime_scrape_failures") != 0 or summary.get("worker_collection_failures") != 0 or summary.get("tidb_probe_failures") != 0 or summary.get("container_probe_failures") != 0 ): raise BaselineError("observation summary is not accepted and complete") assert isinstance(workers, dict) for stem, worker in workers.items(): if ( not isinstance(worker, dict) or worker.get("work_family") != f"runtime.{stem}.v1" or worker.get("identity_valid") is not True or worker.get("required_metrics_complete") is not True or worker.get("collection_failures") != 0 or worker.get("sample_count") != summary.get("sample_count") or not isinstance(worker.get("gauges"), dict) or not isinstance(worker.get("counter_deltas"), dict) ): raise BaselineError(f"observation summary has incomplete {stem} evidence") for name in GAUGE_COSTS: required_number(worker["gauges"], name, stem) for name in EXECUTION_COSTS: required_number(worker["counter_deltas"], name, stem) return worker_stems def validate_identity(identity: dict[str, Any], worker_stems: Sequence[str]) -> None: digest_fields = ( "resolvedBomFileDigest", "desiredBomDigest", "frontendTargetDigest", "effectiveWorkerRolePolicyDigest", ) images = identity.get("images") if ( identity.get("schema") != "nexuscrm.runtime-worker-baseline-identity/v1" or not isinstance(identity.get("deploymentRunId"), str) or not re.fullmatch( r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", identity["deploymentRunId"] ) or not isinstance(identity.get("workerServingSourceRevision"), str) or not re.fullmatch( r"[0-9a-f]{40}", identity["workerServingSourceRevision"] ) or any( not isinstance(identity.get(field), str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", identity[field]) for field in digest_fields ) or not isinstance(images, dict) or set(images) != IMAGE_ROLES or any( not isinstance(image, str) or not re.fullmatch( r"ghcr[.]io/pingcap-inc/[a-z0-9._/-]+@sha256:[0-9a-f]{64}", image, ) for image in images.values() ) or identity.get("runtimeEndpointCount") != 1 or identity.get("workerArtifactCount") != 1 or identity.get("workerRoleCount") != len(worker_stems) ): raise BaselineError("baseline identity is incomplete or malformed") for field in ( "catalogRevision", "securityEpoch", "schemaGeneration", "engineGeneration" ): value = identity.get(field) minimum = 1 if field in {"schemaGeneration", "engineGeneration"} else 0 if not isinstance(value, int) or isinstance(value, bool) or value < minimum: raise BaselineError(f"baseline identity {field} is malformed") def worker_activity(stem: str, worker: dict[str, Any]) -> float: counters = worker.get("counter_deltas") activity = required_number(counters, "processed_delta", stem) if stem == "scheduled-flow": activity += required_number(counters, "scheduled_accepted_delta", stem) activity += required_number( counters, "scheduled_execution_completed_delta", stem ) return activity def validate_semantics( summary: dict[str, Any], scenario: str, baseline_case: str, ) -> None: workers = summary["workers"] if scenario == "quiet": for stem, worker in workers.items(): gauges = worker["gauges"] counters = worker["counter_deltas"] for name in QUIET_ZERO_GAUGES: if required_number(gauges, name, stem) != 0: raise BaselineError(f"quiet run has durable work in {stem}.{name}") quiet_counters = QUIET_ZERO_COUNTERS if stem != "scheduled-flow": quiet_counters = QUIET_ZERO_COUNTERS[:3] for name in quiet_counters: if required_number(counters, name, stem) != 0: raise BaselineError(f"quiet run executed durable work in {stem}.{name}") if window_seconds(summary) < 840: raise BaselineError("quiet run did not retain a full 15-minute observation") elif scenario == "mixed-workload": inactive = [ stem for stem, worker in workers.items() if worker_activity(stem, worker) <= 0 ] if inactive: raise BaselineError("mixed run did not exercise Workers: " + ", ".join(inactive)) def cost_metrics(summary: dict[str, Any], statements: Path) -> dict[str, float]: runtime = summary.get("runtime_peaks") tidb = summary.get("tidb_peaks") if not isinstance(runtime, dict) or not isinstance(tidb, dict): raise BaselineError("Runtime cost evidence is malformed") result = { "runtime.rss_peak_bytes": finite_number( runtime.get("process_resident_memory_bytes"), "Runtime RSS" ), "runtime.heap_peak_bytes": finite_number( runtime.get("go_memstats_heap_inuse_bytes"), "Runtime heap" ), "runtime.primary_pool_open_peak": finite_number( runtime.get("nexuscrm_runtime_db_pool_open_connections"), "Runtime open connections", ), "observation.window_seconds": window_seconds(summary), "tidb.sessions_peak": finite_number(tidb.get("sessions"), "TiDB sessions"), "tidb.running_peak": finite_number(tidb.get("running"), "TiDB running"), "tidb.lock_waits_peak": finite_number(tidb.get("lock_waits"), "TiDB lock waits"), } for stem, worker in summary["workers"].items(): gauges = worker.get("gauges") counters = worker.get("counter_deltas") if not isinstance(gauges, dict) or not isinstance(counters, dict): raise BaselineError(f"{stem} cost evidence is malformed") for name in GAUGE_COSTS: result[f"worker.{stem}.{name}"] = required_number(gauges, name, stem) for name in EXECUTION_COSTS: result[f"worker.{stem}.{name}"] = required_number( counters, name, stem ) result.update(statement_costs(statements)) return result def relative_spread(values: Iterable[float]) -> float: materialized = tuple(values) low = min(materialized) high = max(materialized) if high == 0: return 0.0 if low == 0: return math.inf return (high - low) / low def compare(roots: list[Path]) -> dict[str, Any]: if len(roots) != 3: raise BaselineError("baseline comparison requires exactly three artifacts") runs: list[dict[str, Any]] = [] for root in roots: payload = payload_root(root) identity = load_json(payload / "runtime-worker-baseline-identity.json") context = load_json(payload / "runtime-worker-baseline-context.json") summary = load_json(payload / "observation" / "summary.json") if context.get("schema") != "nexuscrm.runtime-worker-baseline-context/v1": raise BaselineError("baseline context schema is unsupported") scenario = context.get("scenario") baseline_case = context.get("case") validate_scenario_case(scenario, baseline_case) if ( not isinstance(context.get("repetition"), int) or isinstance(context.get("repetition"), bool) or context.get("repetition") not in {1, 2, 3} ): raise BaselineError("baseline scenario or case is unsupported") target_env = summary_target_env(summary) context_target = context.get("targetEnv", "test2") identity_target = identity.get("targetEnv", "test2") if ( context_target != target_env or identity_target != target_env or target_env not in SUPPORTED_TARGET_ENVS ): raise BaselineError( "baseline identity, context, and observation target different Test slots" ) worker_stems = validate_summary(summary) validate_identity(identity, worker_stems) validate_semantics(summary, scenario, baseline_case) runs.append({ "root": str(root), "target_env": target_env, "identity": identity, "context": context, "summary": summary, "metrics": cost_metrics( summary, payload / "observation" / "statement-digests.tsv" ), }) runs.sort(key=lambda run: run["context"].get("repetition", 0)) first_identity = runs[0]["identity"] if any(run["identity"] != first_identity for run in runs[1:]): raise BaselineError("artifacts do not share one frozen deployment/database identity") target_env = runs[0]["target_env"] if any(run["target_env"] != target_env for run in runs[1:]): raise BaselineError("artifacts do not share one selected Test environment") scenario_case = ( runs[0]["context"]["scenario"], runs[0]["context"]["case"] ) if any( (run["context"]["scenario"], run["context"]["case"]) != scenario_case for run in runs[1:] ): raise BaselineError("artifacts do not represent one equivalent scenario and case") repetitions = {run["context"].get("repetition") for run in runs} if repetitions != {1, 2, 3}: raise BaselineError("artifacts must contain baseline repetitions 1, 2, and 3") observer_ids = [run["context"].get("observerRunId") for run in runs] if any(not isinstance(value, str) or not value for value in observer_ids): raise BaselineError("each artifact must own one observer run ID") if len(set(observer_ids)) != 3: raise BaselineError("baseline repetitions must come from distinct observer runs") observer_revisions = [ run["context"].get("observerSourceRevision") for run in runs ] if ( any(not isinstance(value, str) for value in observer_revisions) or len(set(observer_revisions)) != 1 or not re.fullmatch(r"[0-9a-f]{40}", observer_revisions[0]) ): raise BaselineError("baseline repetitions must use one observer revision") scenario = scenario_case[0] workload_ids = [run["context"].get("workloadRunId") for run in runs] if scenario == "quiet": if any(value is not None for value in workload_ids): raise BaselineError("quiet baseline cannot overlap a declared workload run") else: if ( any(not isinstance(value, str) or not value for value in workload_ids) or len(set(workload_ids)) != 3 ): raise BaselineError( "each non-quiet repetition must bind one distinct workload run" ) for run, workload_id in zip(runs, workload_ids): validate_context_workload( run["context"].get("workload"), scenario=scenario, workload_run_id=workload_id, observer_revision=run["context"]["observerSourceRevision"], summary=run["summary"], target_env=target_env, ) workload_inventories = { ( run["context"]["workload"]["caseInventoryDigest"], run["context"]["workload"]["caseCount"], tuple(run["context"]["workload"]["optionalCaseKeys"]), ) for run in runs } if len(workload_inventories) != 1: raise BaselineError( "non-quiet repetitions do not share one workload case inventory" ) names = set(runs[0]["metrics"]) if any(set(run["metrics"]) != names for run in runs[1:]): raise BaselineError("baseline cost ledgers have different metric sets") comparisons = [] unstable = [] for name in sorted(names): values = [run["metrics"][name] for run in runs] spread = relative_spread(values) row = { "metric": name, "values": values, "relativeSpread": None if math.isinf(spread) else spread, "stable": spread <= MAX_RELATIVE_SPREAD, } comparisons.append(row) if not row["stable"]: unstable.append(name) return { "schema": "nexuscrm.runtime-worker-baseline-comparison/v1", "accepted": not unstable, "threshold": MAX_RELATIVE_SPREAD, "targetEnv": target_env, "scenario": scenario, "case": scenario_case[1], "identity": first_identity, "observerRunIds": observer_ids, "workloadRunIds": workload_ids, "comparisons": comparisons, "unstableMetrics": unstable, } def markdown(result: dict[str, Any]) -> str: lines = [ "# Runtime Worker baseline comparison", "", f"- accepted: `{str(result['accepted']).lower()}`", f"- environment: `{result['targetEnv']}`", f"- scenario/case: `{result['scenario']}` / `{result['case']}`", f"- maximum allowed spread: `{result['threshold']:.1%}`", f"- deployment run: `{result['identity']['deploymentRunId']}`", "", "| Metric | Run 1 | Run 2 | Run 3 | Spread | Stable |", "|---|---:|---:|---:|---:|:---:|", ] for row in result["comparisons"]: spread = row["relativeSpread"] spread_text = "infinite" if spread is None else f"{spread:.1%}" values = " | ".join(f"{value:.6g}" for value in row["values"]) lines.append( f"| `{row['metric']}` | {values} | {spread_text} | " f"{'yes' if row['stable'] else 'no'} |" ) return "\n".join(lines) + "\n" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run", action="append", required=True, type=Path) parser.add_argument("--json-output", type=Path) parser.add_argument("--markdown-output", type=Path) return parser.parse_args() def main() -> int: args = parse_args() try: result = compare(args.run) except BaselineError as exc: print(f"runtime worker baseline rejected: {exc}", file=sys.stderr) return 2 rendered = markdown(result) if args.json_output: args.json_output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") if args.markdown_output: args.markdown_output.write_text(rendered) else: print(rendered, end="") return 0 if result["accepted"] else 1 if __name__ == "__main__": raise SystemExit(main())