from __future__ import annotations import copy import json import math import sys from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from backend.sql_catalog import server # noqa: E402 from scripts.import_clinic_health_report import ( # noqa: E402 DETECTOR_VERSION, SOURCE, build_findings, source_run_id, ) REPORT_PATH = ROOT / "data" / "mock" / "TiDB Clinic Health Check Report.json" DERIVED_BY = "clinic-report-demo-builder" SCALABLE_METRIC_KEYS = frozenset( { "slow_count", "total_latency_s", "max_latency_s", "total_keys", "process_keys", "request_count", "sum_ru", } ) def parse_time(value: str) -> datetime: return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) def iso_time(value: datetime) -> str: return value.replace(microsecond=0).isoformat().replace("+00:00", "Z") def shift_time(value: str | None, delta: timedelta) -> str | None: return iso_time(parse_time(value) + delta) if value else None @contextmanager def domain_time(value: datetime): original = server.now_iso server.now_iso = lambda: iso_time(value) try: yield finally: server.now_iso = original def reset_demo_database() -> None: with server.get_conn() as conn: for table in ( "timeline_events", "relations", "records", "treatments", "finding_symptoms", "findings", "finding_deletions", "finding_runs", "catalog_items", ): conn.execute(f"DELETE FROM {table}") def cluster_scopes(report: dict[str, Any]) -> list[dict[str, str]]: cluster_id = str(report["clusterId"]) return [ {"cluster_id": cluster_id, "profile": "production"}, {"cluster_id": f"{cluster_id}-replica-a", "profile": "replica-a"}, {"cluster_id": f"{cluster_id}-replica-b", "profile": "replica-b"}, ] def run_endpoints(days: int) -> list[datetime]: now = datetime.now(timezone.utc) latest = now.replace(hour=6, minute=0, second=0, microsecond=0) if latest + timedelta(minutes=5) > now: latest -= timedelta(days=1) return [latest - timedelta(days=days - index - 1) for index in range(days)] def active_templates( templates: list[dict[str, Any]], day_index: int, day_count: int, cluster_index: int, ) -> list[dict[str, Any]]: starts = (24, 18, 14) floors = (8, 6, 4) progress = day_index / max(1, day_count - 1) active_count = round(starts[cluster_index] - (starts[cluster_index] - floors[cluster_index]) * progress) base_count = min(18, len(templates)) unlock_days = (4, 9, 14, 19, 23, 26) unlocked_count = sum(day_index >= unlock_day for unlock_day in unlock_days) unlocked = templates[base_count : base_count + unlocked_count] stable = templates[:base_count] selected = [*reversed(unlocked), *stable] return selected[: min(active_count, len(selected))] def demo_intake_template(templates: list[dict[str, Any]]) -> dict[str, Any]: candidates = [ template for template in templates if template.get("detector_code") == "clinic.slowlog_top_digest" and server.finding_severity(template.get("symptoms") or []) == "critical" and template.get("metrics") ] if not candidates: raise RuntimeError("The Clinic report does not contain a Critical slow-log Finding") return sorted(candidates, key=lambda template: str(template["sql_digest"]))[0] def metric_scale_factor( template: dict[str, Any], day_index: int, day_count: int, cluster_index: int ) -> float: digest = str(template["sql_digest"]) seed = int(digest[:8], 16) if len(digest) >= 8 else sum(ord(value) for value in digest) progress = day_index / max(1, day_count - 1) trend = 1.16 - 0.24 * progress wave = math.sin((day_index + seed % 11) * 0.78) * 0.07 cluster_bias = 1 + cluster_index * 0.06 return round(max(0.65, (trend + wave) * cluster_bias), 4) def scale_metrics(metrics: dict[str, Any], factor: float) -> dict[str, Any]: result = copy.deepcopy(metrics) for key in SCALABLE_METRIC_KEYS: value = result.get(key) if isinstance(value, bool) or not isinstance(value, (int, float)): continue if isinstance(value, int): result[key] = max(0, int(round(value * factor))) else: result[key] = round(max(0.0, float(value) * factor), 3) return result def create_run( report: dict[str, Any], cluster: dict[str, str], run_end: datetime, run_id: str, *, status: str, completed_detectors: int, derivation: str, ) -> dict[str, Any]: run_start = run_end - timedelta(days=1) response_status, response = server.create_finding_run( { "source": SOURCE, "source_type": "inspection", "source_run_id": run_id, "org_id": report["dbaasOrgId"], "cluster_id": cluster["cluster_id"], "time_window_start": iso_time(run_start), "time_window_end": iso_time(run_end), "status": status, "detector_version": DETECTOR_VERSION, "expected_detector_count": 3, "completed_detector_count": completed_detectors, "metadata": { "strategy_uuid": report.get("strategyUuid"), "report_created_at": report.get("createdAt"), "source_report": str(REPORT_PATH.relative_to(ROOT)), "source_cluster_id": report.get("clusterId"), "fixture_derivation": derivation, "profile": cluster["profile"], }, } ) if response_status not in {200, 201}: raise RuntimeError(response) return response["run"] def derived_finding( template: dict[str, Any], report: dict[str, Any], cluster: dict[str, str], run_id: str, run_end: datetime, *, exact_report: bool, metric_scale: float = 1.0, ) -> dict[str, Any]: payload = copy.deepcopy(template) original_end = parse_time(str(report["timeRangeAEnd"])) delta = run_end - original_end detector = str(payload.get("detector_code") or "clinic.unknown") digest = str(payload["sql_digest"]) payload.update( { "source_type": "inspection", "source_run_id": run_id, "source_event_id": f"{SOURCE}:{run_id}:{detector}:{digest}", "cluster_id": cluster["cluster_id"], "time_window_start": shift_time(payload.get("time_window_start"), delta), "time_window_end": shift_time(payload.get("time_window_end"), delta), "observed_at": shift_time(payload.get("observed_at"), delta), "created_by": DERIVED_BY, } ) dimensions = copy.deepcopy(payload.get("dimensions") or {}) dimensions["fixture_derivation"] = { "mode": "none" if exact_report else "time_scope_and_metric_scale", "source_report": str(REPORT_PATH.relative_to(ROOT)), "source_cluster_id": report.get("clusterId"), "source_time_window_end": report.get("timeRangeAEnd"), "metric_scale_factor": metric_scale, } payload["dimensions"] = dimensions if not exact_report and metric_scale != 1: payload["metrics"] = scale_metrics(payload.get("metrics") or {}, metric_scale) raw_payload = payload.get("raw_payload") or {} payload["raw_payload"] = { "report": raw_payload.get("report"), "check": raw_payload.get("check"), "fixture_derivation": dimensions["fixture_derivation"], } return payload def set_fixture_ingestion_time(finding_id: str, run_id: str, run_end: datetime) -> None: ingested_at = iso_time(run_end + timedelta(minutes=5)) with server.get_conn() as conn: conn.execute( "UPDATE findings SET ingested_at = ?, created_at = ? WHERE id = ?", (ingested_at, ingested_at, finding_id), ) conn.execute( "UPDATE finding_runs SET ingested_at = ? WHERE id = ?", (ingested_at, run_id), ) def seed_findings(report: dict[str, Any]) -> dict[str, Any]: templates = sorted( build_findings(report), key=lambda item: ( -server.SEVERITY_RANK[server.finding_severity(item["symptoms"])], str(item.get("detector_code")), str(item.get("sql_digest")), ), ) reserved_template = demo_intake_template(templates) reserved_digest = str(reserved_template["sql_digest"]) scopes = cluster_scopes(report) endpoints = run_endpoints(28) actual_report_end = parse_time(str(report["timeRangeAEnd"])) created_findings = 0 created_runs = 0 for day_index, run_end in enumerate(endpoints): for cluster_index, cluster in enumerate(scopes): exact_report = ( cluster_index == 0 and run_end.date() == actual_report_end.date() ) effective_run_end = actual_report_end if exact_report else run_end run_id = ( source_run_id(report) if exact_report else f"{report['strategyUuid']}:derived:{run_end.date()}:{cluster['cluster_id']}" ) failed = cluster_index == 2 and day_index == 13 partial = cluster_index == 1 and day_index == 8 zero_finding = cluster_index == 2 and day_index == 22 status = "failed" if failed else "partial" if partial else "success" completed_detectors = 0 if failed else 2 if partial else 3 run = create_run( report, cluster, effective_run_end, run_id, status=status, completed_detectors=completed_detectors, derivation="none" if exact_report else "time_and_scope_shift", ) created_runs += 1 candidates = ( templates if exact_report else active_templates(templates, day_index, len(endpoints), cluster_index) ) if failed or zero_finding: candidates = [] if cluster_index == 0: candidates = [ template for template in candidates if str(template["sql_digest"]) != reserved_digest ] payloads = [ derived_finding( template, report, cluster, run_id, effective_run_end, exact_report=exact_report, metric_scale=( 1.0 if exact_report else metric_scale_factor( template, day_index, len(endpoints), cluster_index ) ), ) for template in candidates ] if payloads: status_code, response = server.create_finding_batch( { "source": SOURCE, "source_type": "inspection", "source_batch_id": f"fixture:{run_id}", "findings": payloads, "created_by": DERIVED_BY, } ) if status_code not in {200, 201} or response["summary"]["rejected"]: raise RuntimeError(response) created_findings += response["summary"]["created"] for item in response["items"]: if item["status"] not in {"created", "duplicate"}: continue set_fixture_ingestion_time( item["finding_id"], run["id"], effective_run_end ) with server.get_conn() as conn: conn.execute( """ UPDATE catalog_items SET created_at = COALESCE(( SELECT MIN(f.created_at) FROM findings f WHERE f.catalog_item_id = catalog_items.id ), created_at), updated_at = COALESCE(( SELECT MAX(f.created_at) FROM findings f WHERE f.catalog_item_id = catalog_items.id ), updated_at) """ ) return { "runs": created_runs, "findings": created_findings, "clusters": [scope["cluster_id"] for scope in scopes], "templates": len(templates), "reserved_live_intake": { "cluster_id": str(report["clusterId"]), "sql_digest": reserved_digest, "severity": server.finding_severity(reserved_template["symptoms"]), "symptoms": [symptom["code"] for symptom in reserved_template["symptoms"]], }, } def ensure_record( target_type: str, target_id: str, record_type: str, title: str, *, state: str = "ready", visibility: str = "internal", fields: dict[str, Any] | None = None, body: str = "", occurred_at: datetime | None = None, ) -> dict[str, Any]: with domain_time(occurred_at or datetime.now(timezone.utc)): status, result = server.create_record( target_type, target_id, { "record_type": record_type, "title": title, "review_status": state, "visibility": visibility, "structured_fields": fields or {}, "body_markdown": body, "created_by": DERIVED_BY, }, ) if status != 201: raise RuntimeError(result) return result["record"] def create_treatment(item_id: str, title: str, occurred_at: datetime) -> dict[str, Any]: with domain_time(occurred_at): status, response = server.create_treatment( { "title": title, "objective": "Reduce the operational impact of the Clinic-identified SQL.", "success_criteria": "A comparable Clinic observation confirms material improvement without workload regression.", "owner": "sql-governance-demo", "created_by": DERIVED_BY, }, item_id, ) if status != 201: raise RuntimeError(response) return response["treatment"] def catalog_candidates(cluster_id: str) -> list[dict[str, Any]]: with server.get_conn() as conn: return [ server.row_to_catalog_item(row) for row in conn.execute( """ SELECT * FROM catalog_items WHERE cluster_id = ? ORDER BY CASE severity WHEN 'critical' THEN 4 WHEN 'high' THEN 3 WHEN 'medium' THEN 2 ELSE 1 END DESC, first_seen_at LIMIT 6 """, (cluster_id,), ).fetchall() ] def seed_governance(report: dict[str, Any]) -> dict[str, Any]: items = catalog_candidates(str(report["clusterId"])) if len(items) < 6: raise RuntimeError("The Clinic report did not produce enough Catalog Items") anchor = datetime.now(timezone.utc).replace(microsecond=0) def move(treatment_id: str, stage: str, occurred_at: datetime) -> None: with domain_time(occurred_at): status, response = server.transition_treatment( treatment_id, {"stage": stage, "actor": DERIVED_BY} ) if status != 200: raise RuntimeError(response) watch_item = items[0] watch_record = ensure_record( "catalog_item", watch_item["id"], "triage_record", "Continue observing the Clinic signal", fields={"conclusion_type": "watch", "confidence": "medium", "source": SOURCE}, body="## Decision\n\nKeep the SQL under observation until another comparable Clinic window is available.", occurred_at=anchor - timedelta(days=1), ) with domain_time(anchor - timedelta(days=1)): status, response = server.triage_catalog_item( watch_item["id"], {"action": "watch", "record_id": watch_record["id"], "actor": DERIVED_BY}, ) if status != 200: raise RuntimeError(response) improved_item = items[1] improved_treatment = create_treatment( improved_item["id"], "Reduce latency for Clinic-identified SQL", anchor - timedelta(days=11), ) ensure_record( "treatment", improved_treatment["id"], "analysis_record", "Slow-log evidence confirmed", fields={"conclusion_type": "query_plan_or_index", "confidence": "high"}, body="## Diagnosis\n\nClinic slow-log evidence shows sustained latency and key-processing cost for this digest.", occurred_at=anchor - timedelta(days=10, hours=12), ) move(improved_treatment["id"], "planning", anchor - timedelta(days=10)) ensure_record( "treatment", improved_treatment["id"], "proposal_record", "Optimize access path and validate", state="reviewed", visibility="customer_visible", fields={"action_type": "sql_or_index_change", "risk_level": "medium"}, body="## Proposal\n\nValidate the access path in staging, then compare the same Clinic metrics after rollout.", occurred_at=anchor - timedelta(days=9), ) move(improved_treatment["id"], "execution", anchor - timedelta(days=8)) observation = ensure_record( "treatment", improved_treatment["id"], "observation_record", "Comparable Clinic window shows improvement", state="reviewed", fields={ "baseline_source": str(REPORT_PATH.relative_to(ROOT)), "traffic_comparable": True, "conclusion_type": "improved", "confidence": "high", "metric_change": {"total_latency_ratio": 0.24, "slow_count_ratio": 0.31}, }, body="## Result\n\nThe derived follow-up window retains the report's metric schema and shows a material reduction in latency and occurrence count.", occurred_at=anchor - timedelta(days=5, hours=12), ) with domain_time(anchor - timedelta(days=5)): status, response = server.start_observation( improved_treatment["id"], { "record_id": observation["id"], "observation_due_at": iso_time(anchor - timedelta(days=2)), "actor": DERIVED_BY, }, ) if status != 200: raise RuntimeError(response) with domain_time(anchor - timedelta(days=3)): status, response = server.complete_treatment( improved_treatment["id"], {"outcome": "improved", "actor": DERIVED_BY} ) if status != 200: raise RuntimeError(response) observation_item = items[2] observation_treatment = create_treatment( observation_item["id"], "Validate remediation against the next Clinic run", anchor - timedelta(days=6), ) ensure_record( "treatment", observation_treatment["id"], "analysis_record", "Clinic signal requires a comparable follow-up", fields={"conclusion_type": "needs_observation", "confidence": "medium"}, body="The current evidence is actionable, but the result must be confirmed against a comparable inspection window.", occurred_at=anchor - timedelta(days=5, hours=12), ) move(observation_treatment["id"], "planning", anchor - timedelta(days=5)) ensure_record( "treatment", observation_treatment["id"], "proposal_record", "Reduce repeated key processing before the next Clinic run", state="reviewed", visibility="customer_visible", fields={"action_type": "sql_or_index_change", "risk_level": "low"}, body="The recommendation is derived from the recurring key-processing signal in the Clinic report.", occurred_at=anchor - timedelta(days=4), ) move(observation_treatment["id"], "execution", anchor - timedelta(days=3)) pending_observation = ensure_record( "treatment", observation_treatment["id"], "observation_record", "Awaiting the next complete Clinic run", state="draft", fields={"traffic_comparable": False, "conclusion_type": "", "confidence": "medium"}, body="Observation remains pending until a complete run with the same detector version is available.", occurred_at=anchor - timedelta(days=2, hours=2), ) with domain_time(anchor - timedelta(days=2)): status, response = server.start_observation( observation_treatment["id"], { "record_id": pending_observation["id"], "observation_due_at": iso_time(anchor + timedelta(days=1)), "actor": DERIVED_BY, }, ) if status != 200: raise RuntimeError(response) planning_item = items[3] planning_treatment = create_treatment( planning_item["id"], "Design remediation for recurring scan-heavy SQL", anchor - timedelta(days=12), ) ensure_record( "treatment", planning_treatment["id"], "analysis_record", "Recurring scan cost confirmed", fields={"conclusion_type": "query_plan_or_index", "confidence": "medium"}, body="The same Clinic-derived digest remains present across multiple report windows.", occurred_at=anchor - timedelta(days=11, hours=12), ) move(planning_treatment["id"], "planning", anchor - timedelta(days=10)) alignment_item = items[4] alignment_treatment = create_treatment( alignment_item["id"], "Align rollout for a high-impact SQL change", anchor - timedelta(days=5), ) ensure_record( "treatment", alignment_treatment["id"], "analysis_record", "High-impact SQL requires coordinated rollout", fields={"conclusion_type": "query_plan_or_index", "confidence": "medium"}, body="Clinic evidence supports intervention, but rollout ownership and timing still need agreement.", occurred_at=anchor - timedelta(days=4, hours=12), ) move(alignment_treatment["id"], "planning", anchor - timedelta(days=4)) ensure_record( "treatment", alignment_treatment["id"], "proposal_record", "Coordinate index rollout and application validation", fields={"action_type": "sql_or_index_change", "risk_level": "medium"}, body="The draft remains internal until the delivery window and owners are agreed.", occurred_at=anchor - timedelta(days=3), ) move(alignment_treatment["id"], "alignment", anchor - timedelta(days=2, hours=12)) assessment_item = items[5] assessment_treatment = create_treatment( assessment_item["id"], "Assess a newly prioritized Clinic SQL signal", anchor - timedelta(days=1), ) return { "watching_catalog_item": watch_item["id"], "completed_treatment": improved_treatment["id"], "observation_treatment": observation_treatment["id"], "planning_treatment": planning_treatment["id"], "alignment_treatment": alignment_treatment["id"], "assessment_treatment": assessment_treatment["id"], } def prepare_demo() -> dict[str, Any]: report = json.loads(REPORT_PATH.read_text(encoding="utf-8")) server.init_db() reset_demo_database() findings = seed_findings(report) governance = seed_governance(report) return { "source_report": str(REPORT_PATH), "findings": findings, "governance": governance, } def main() -> None: print(json.dumps(prepare_demo(), ensure_ascii=False, indent=2)) if __name__ == "__main__": main()