"""Versioned, privacy-minimised Dashboard v5 bundle contract.""" from __future__ import annotations import json import math import re from datetime import date, datetime, timedelta from statistics import median from typing import Any from dashboard_v5.lab_registry import EXACT_LAB_NUMBER, PUBLIC_LAB_PAIRS from dashboard_v5.metric_registry import public_explorer, public_registry SCHEMA_VERSION = "health_dashboard.bundle.v2" ALLOWED_TOP_LEVEL = { "schema_version", "generated_at", "timezone", "today", "metrics", "explorer", "series", "events", "medication", "labs", "nutrition", "tasks", "notices", "freshness", "correlations", } MAX_BUNDLE_BYTES = 512_000 MAX_METRICS = 32 MAX_SERIES_POINTS = 3_660 MAX_EVENTS = 1_000 MAX_LABS = 100 MAX_NUTRITION_DAYS = 1_000 MAX_REVIEW_ITEMS = 100 MAX_TASKS = 20 MAX_PRESETS = 16 MAX_CORRELATIONS = 4_096 PATH_LIKE = re.compile(r"^(?:/|~/|[A-Za-z]:[\\/]|file:|https?://)|(?:^|/)\.hermes(?:/|$)", re.I) def _bounded_list(value: Any, path: str, maximum: int) -> list[Any]: if not isinstance(value, list): raise TypeError(f"{path} must be a list") if len(value) > maximum: raise ValueError(f"too many items at {path}") return value def _exact(mapping: Any, keys: set[str], path: str) -> dict[str, Any]: if not isinstance(mapping, dict) or set(mapping) != keys: raise ValueError(f"invalid fields at {path}") return mapping def _bounded_text(value: Any, path: str, maximum: int = 200, *, allow_empty: bool = False) -> str: if not isinstance(value, str): raise TypeError(f"non-string at {path}") if value != value.strip() or any(ord(char) < 32 for char in value): raise ValueError(f"unsafe whitespace at {path}") if (not value and not allow_empty) or len(value) > maximum or PATH_LIKE.search(value): raise ValueError(f"unsafe text at {path}") return value def _iso_day(value: Any, path: str) -> str: text = _bounded_text(value, path, 10) if date.fromisoformat(text).isoformat() != text: raise ValueError(f"invalid day at {path}") return text def _finite_optional(value: Any, path: str) -> None: if value is None: return if type(value) not in {int, float} or not math.isfinite(value): raise ValueError(f"non-finite/non-numeric value at {path}") def _validate_metric(entry: Any, index: int) -> str: path = f"metrics[{index}]" metric = _exact(entry, { "id", "label", "unit", "unit_family", "raw_overlay_group", "group", "aggregation", "baseline_days", "baseline_min_observations", "index_mode", "correlation_id", "decimals", }, path) metric_id = _bounded_text(metric["id"], f"{path}.id", 80) if not re.fullmatch(r"[a-z][a-z0-9_.]*", metric_id): raise ValueError(f"invalid metric id at {path}") for field, maximum in (("label", 80), ("unit", 24), ("unit_family", 40), ("raw_overlay_group", 40), ("correlation_id", 80), ("group", 40), ("aggregation", 16)): _bounded_text(metric[field], f"{path}.{field}", maximum) if not re.fullmatch(r"[a-z][a-z0-9_:]*", metric["correlation_id"]): raise ValueError(f"invalid correlation id at {path}") if metric["index_mode"] not in {"ratio_100", "unsupported"}: raise ValueError(f"invalid index mode at {path}") if type(metric["baseline_days"]) is not int or not 1 <= metric["baseline_days"] <= 365: raise ValueError(f"invalid baseline at {path}") if type(metric["baseline_min_observations"]) is not int or not 1 <= metric["baseline_min_observations"] <= metric["baseline_days"]: raise ValueError(f"invalid baseline minimum at {path}") if type(metric["decimals"]) is not int or not 0 <= metric["decimals"] <= 4: raise ValueError(f"invalid decimals at {path}") return metric_id def validate_bundle(bundle: dict[str, Any]) -> None: _exact(bundle, ALLOWED_TOP_LEVEL, "bundle") if bundle["schema_version"] != SCHEMA_VERSION: raise ValueError("unsupported schema version") datetime.fromisoformat(_bounded_text(bundle["generated_at"], "generated_at", 40).replace("Z", "+00:00")) if bundle["timezone"] != "Europe/Zurich": raise ValueError("unexpected timezone") today = _iso_day(bundle["today"], "today") metrics = _bounded_list(bundle["metrics"], "metrics", MAX_METRICS) metric_ids = {_validate_metric(entry, index) for index, entry in enumerate(metrics)} if len(metric_ids) != len(metrics): raise ValueError("duplicate metric id") if metrics != public_registry(): raise ValueError("metric registry differs from the canonical server configuration") explorer = _exact(bundle["explorer"], {"modes", "presets"}, "explorer") modes = _exact(explorer["modes"], {"raw", "baseline_index"}, "explorer.modes") if modes["raw"] != {"max_series": 2, "compatibility": "same_raw_overlay_group"}: raise ValueError("invalid raw Explorer contract") if modes["baseline_index"] != {"max_series": 4, "reference": 100}: raise ValueError("invalid baseline Explorer contract") metric_by_id = {entry["id"]: entry for entry in metrics} preset_ids: set[str] = set() presets = _bounded_list(explorer["presets"], "explorer.presets", MAX_PRESETS) for index, entry in enumerate(presets): path = f"explorer.presets[{index}]" preset = _exact(entry, {"id", "label", "mode", "metric_ids"}, path) preset_id = _bounded_text(preset["id"], f"{path}.id", 60) _bounded_text(preset["label"], f"{path}.label", 80) if preset_id in preset_ids or preset["mode"] not in modes: raise ValueError(f"invalid preset at {path}") preset_ids.add(preset_id) selected = preset["metric_ids"] if not isinstance(selected, list) or not selected or len(selected) != len(set(selected)): raise ValueError(f"invalid preset metrics at {path}") maximum = modes[preset["mode"]]["max_series"] if len(selected) > maximum or any(metric_id not in metric_ids for metric_id in selected): raise ValueError(f"preset exceeds allowlist or limit at {path}") if preset["mode"] == "raw": groups = {metric_by_id[metric_id]["raw_overlay_group"] for metric_id in selected} if len(groups) != 1: raise ValueError(f"incompatible raw preset at {path}") elif any(metric_by_id[metric_id]["index_mode"] != "ratio_100" for metric_id in selected): raise ValueError(f"unsupported index preset at {path}") if explorer != public_explorer(): raise ValueError("Explorer registry differs from the canonical server configuration") if not isinstance(bundle["series"], dict) or set(bundle["series"]) != metric_ids: raise ValueError("series must exactly match the metric registry") for metric_id, points_value in bundle["series"].items(): points = _bounded_list(points_value, f"series.{metric_id}", MAX_SERIES_POINTS) metric = metric_by_id[metric_id] previous_day: str | None = None observed: list[tuple[date, float]] = [] for index, entry in enumerate(points): path = f"series.{metric_id}[{index}]" point = _exact(entry, {"date", "value", "quality", "baseline"}, path) day = _iso_day(point["date"], f"{path}.date") if day > today: raise ValueError(f"future measurement day at {path}") if previous_day is not None and day <= previous_day: raise ValueError(f"series dates must be strictly ascending at {path}") previous_day = day _finite_optional(point["value"], f"{path}.value") _finite_optional(point["baseline"], f"{path}.baseline") quality = point["quality"] if quality not in {"complete", "observed", "incomplete"}: raise ValueError(f"invalid quality at {path}") if (quality == "incomplete") != (point["value"] is None): raise ValueError(f"inconsistent value/quality at {path}") current_day = date.fromisoformat(day) start = current_day - timedelta(days=metric["baseline_days"]) prior = [value for prior_day, value in observed if start <= prior_day < current_day] expected_baseline = round(median(prior), 3) if len(prior) >= metric["baseline_min_observations"] else None if expected_baseline is None: if point["baseline"] is not None: raise ValueError(f"premature baseline at {path}") elif point["baseline"] is None or not math.isclose(float(point["baseline"]), expected_baseline, rel_tol=0, abs_tol=1e-9): raise ValueError(f"incorrect baseline at {path}") if point["value"] is not None: observed.append((current_day, float(point["value"]))) events = _bounded_list(bundle["events"], "events", MAX_EVENTS) for index, entry in enumerate(events): event = _exact(entry, {"date", "type", "label"}, f"events[{index}]") event_day = _iso_day(event["date"], f"events[{index}].date") if event_day > today or event["type"] != "medication_administered": raise ValueError(f"invalid event day or type at events[{index}]") _bounded_text(event["type"], f"events[{index}].type", 40) _bounded_text(event["label"], f"events[{index}].label", 120) medication = _exact(bundle["medication"], {"last_administered", "next_planned"}, "medication") for field, keys in ( ("last_administered", {"date", "name", "dose"}), ("next_planned", {"date", "name", "dose", "status"}), ): item = medication[field] if item is None: continue item = _exact(item, keys, f"medication.{field}") medication_day = _iso_day(item["date"], f"medication.{field}.date") if (field == "last_administered" and medication_day > today) or (field == "next_planned" and medication_day <= today): raise ValueError(f"invalid medication date at medication.{field}") _bounded_text(item["name"], f"medication.{field}.name", 120) _bounded_text(item["dose"], f"medication.{field}.dose", 40, allow_empty=True) if field == "next_planned" and item["status"] != "planned": raise ValueError("invalid planned medication status") labs = _bounded_list(bundle["labs"], "labs", MAX_LABS) for index, entry in enumerate(labs): path = f"labs[{index}]" lab = _exact(entry, {"parameter", "value", "unit", "date", "reference_min", "reference_max", "quality", "source_type"}, path) for field, maximum in (("parameter", 100), ("value", 32), ("unit", 24), ("source_type", 32)): _bounded_text(lab[field], f"{path}.{field}", maximum) for field in ("reference_min", "reference_max"): if lab[field] is not None: reference_value = str(lab[field]).strip() _bounded_text(reference_value, f"{path}.{field}", 32) if not EXACT_LAB_NUMBER.fullmatch(reference_value): raise ValueError(f"non-numeric lab reference at {path}.{field}") if _iso_day(lab["date"], f"{path}.date") > today: raise ValueError(f"future lab day at {path}") if not EXACT_LAB_NUMBER.fullmatch(lab["value"]): raise ValueError(f"non-numeric verified lab value at {path}") if lab["quality"] != "verified_original" or lab["source_type"] != "scanned_original": raise ValueError(f"invalid lab provenance at {path}") if (lab["parameter"], lab["unit"]) not in PUBLIC_LAB_PAIRS: raise ValueError(f"non-allowlisted lab parameter/unit at {path}") nutrition = _exact(bundle["nutrition"], {"daily", "mapping_review", "mapping_open_count"}, "nutrition") if type(nutrition["mapping_open_count"]) is not int or nutrition["mapping_open_count"] < 0: raise ValueError("invalid mapping count") daily_rows = _bounded_list(nutrition["daily"], "nutrition.daily", MAX_NUTRITION_DAYS) previous_nutrition_day: str | None = None for index, entry in enumerate(daily_rows): path = f"nutrition.daily[{index}]" row = _exact(entry, {"date", "kcal", "item_count", "histamine_score", "histamine_status", "unknown_items", "quality"}, path) nutrition_day = _iso_day(row["date"], f"{path}.date") if nutrition_day > today or (previous_nutrition_day is not None and nutrition_day <= previous_nutrition_day): raise ValueError(f"invalid nutrition date order at {path}") previous_nutrition_day = nutrition_day for field in ("kcal", "histamine_score"): _finite_optional(row[field], f"{path}.{field}") for field in ("item_count", "unknown_items"): if row[field] is not None and (type(row[field]) is not int or row[field] < 0): raise ValueError(f"invalid count at {path}.{field}") if row["histamine_status"] not in {"green", "yellow", "orange", "red", "unknown"}: raise ValueError(f"invalid histamine status at {path}") if row["quality"] not in {"complete", "incomplete"}: raise ValueError(f"invalid nutrition quality at {path}") if row["quality"] == "complete": if row["histamine_score"] is None or row["histamine_status"] == "unknown" or row["unknown_items"] != 0: raise ValueError(f"inconsistent complete nutrition row at {path}") elif row["histamine_score"] is not None or row["histamine_status"] != "unknown": raise ValueError(f"inconsistent incomplete nutrition row at {path}") review_rows = _bounded_list(nutrition["mapping_review"], "nutrition.mapping_review", MAX_REVIEW_ITEMS) for index, entry in enumerate(review_rows): path = f"nutrition.mapping_review[{index}]" row = _exact(entry, {"name", "occurrences", "reason"}, path) _bounded_text(row["name"], f"{path}.name", 160) _bounded_text(row["reason"], f"{path}.reason", 120) if type(row["occurrences"]) is not int or row["occurrences"] < 0: raise ValueError(f"invalid occurrences at {path}") tasks = _bounded_list(bundle["tasks"], "tasks", min(MAX_TASKS, 3)) for index, entry in enumerate(tasks): allowed = {"id", "label", "action", "count"} if not isinstance(entry, dict) or not {"id", "label", "action"} <= set(entry) <= allowed: raise ValueError(f"invalid task at {index}") for field in ("id", "label", "action"): _bounded_text(entry[field], f"tasks[{index}].{field}", 120) if "count" in entry and (type(entry["count"]) is not int or entry["count"] < 0): raise ValueError(f"invalid task count at {index}") if bundle["notices"] != [] or bundle["freshness"] != []: raise ValueError("notices/freshness are reserved in bundle v2") import multimodal_correlations as engine correlations = _exact(bundle["correlations"], {"dimensions", "results"}, "correlations") dimensions = _exact(correlations["dimensions"], {"lag_days", "medication_phases", "lag_direction"}, "correlations.dimensions") if dimensions != { "lag_days": list(engine.LAGS), "medication_phases": sorted(engine.ALLOWED_PHASES), "lag_direction": "predictor_d_to_target_d_plus_lag", }: raise ValueError("invalid correlation dimensions") result_rows = _bounded_list(correlations["results"], "correlations.results", MAX_CORRELATIONS) previous_order: tuple[str, str, int, str] | None = None base_flags = ["complete_case", "phase_stratified", "7d_block_permutation", "no_causality"] for index, entry in enumerate(result_rows): path = f"correlations.results[{index}]" row = _exact(entry, {"id", "dimensions", "sample", "statistics", "status", "method", "quality_flags"}, path) identity = _exact(row["id"], {"predictor", "target"}, f"{path}.id") dims = _exact(row["dimensions"], {"lag_days", "medication_phase"}, f"{path}.dimensions") sample = _exact(row["sample"], {"eligible_target_days", "expected_target_days", "observed_pairs", "missing_pairs", "target_coverage"}, f"{path}.sample") statistics = _exact(row["statistics"], {"rho", "p_value", "q_value"}, f"{path}.statistics") if identity["predictor"] not in engine.ALLOWED_PREDICTORS or identity["target"] not in engine.ALLOWED_TARGETS: raise ValueError(f"unallowlisted correlation identity at {path}") if dims["lag_days"] not in engine.LAGS or dims["medication_phase"] not in engine.ALLOWED_PHASES: raise ValueError(f"invalid correlation dimensions at {path}") if row["status"] not in engine.ALLOWED_STATUSES or row["method"] != engine.METHOD: raise ValueError(f"invalid correlation status or method at {path}") order = (identity["predictor"], identity["target"], dims["lag_days"], dims["medication_phase"]) if previous_order is not None and order <= previous_order: raise ValueError("correlation results are not uniquely and deterministically ordered") previous_order = order counts = [sample[key] for key in ("observed_pairs", "eligible_target_days", "expected_target_days", "missing_pairs")] if any(type(value) is not int or value < 0 for value in counts): raise ValueError(f"invalid correlation counts at {path}") n, eligible, expected, missing = counts if expected < 1 or not n <= eligible <= expected or missing != eligible - n: raise ValueError(f"inconsistent correlation counts at {path}") coverage = sample["target_coverage"] _finite_optional(coverage, f"{path}.sample.target_coverage") if not 0 <= coverage <= 1 or not math.isclose(coverage, eligible / expected, rel_tol=0, abs_tol=5e-7): raise ValueError(f"inconsistent target coverage at {path}") for field, lower, upper in (("rho", -1, 1), ("p_value", 0, 1), ("q_value", 0, 1)): value = statistics[field] _finite_optional(value, f"{path}.statistics.{field}") if value is not None and not lower <= value <= upper: raise ValueError(f"out-of-range statistic at {path}.{field}") if row["status"] == "computed": if any(statistics[field] is None for field in statistics) or coverage < engine.MIN_TARGET_COVERAGE or n < engine.MIN_PAIRS: raise ValueError(f"inconsistent computed result at {path}") expected_flags = base_flags elif row["status"] == "insufficient_coverage": if any(statistics[field] is not None for field in statistics) or coverage >= engine.MIN_TARGET_COVERAGE: raise ValueError(f"inconsistent low-coverage result at {path}") expected_flags = base_flags + ["low_target_coverage"] else: if any(statistics[field] is not None for field in statistics) or coverage < engine.MIN_TARGET_COVERAGE: raise ValueError(f"inconsistent low-n result at {path}") expected_flags = base_flags + ["low_n_or_variance"] if row["quality_flags"] != expected_flags: raise ValueError(f"invalid quality flags at {path}") def safe_json(bundle: dict[str, Any]) -> str: validate_bundle(bundle) payload = json.dumps(bundle, ensure_ascii=False, separators=(",", ":")).replace("<", "\\u003c") if len(payload.encode("utf-8")) > MAX_BUNDLE_BYTES: raise ValueError("bundle exceeds serialized size budget") return payload