def build_run_summary(*, runtime_context: dict, contract: dict | None = None, source_profile: dict | None = None, output_profile: dict | None = None, schema_drift_result: dict | None = None, incremental_safety_result: dict | None = None, quality_result: dict | None = None, contract_validation_result: dict | None = None, lineage_summary: dict | None = None, notes: list[str] | None = None) -> dict:
"""Build a handover-friendly summary for one data product run.
Parameters
----------
runtime_context : dict
Runtime identifiers and environment metadata for the run.
contract : dict | None
Optional normalized data-product contract used for validation and targets.
Returns
-------
dict
Consolidated summary payload used for lineage and handover documentation.
"""
sections = {"purpose": (contract or {}).get("dataset", {}).get("purpose"), "source_profile": source_profile, "output_profile": output_profile, "schema_drift": schema_drift_result, "incremental_safety": incremental_safety_result, "quality": quality_result, "contracts": contract_validation_result, "lineage": lineage_summary, "notes": notes or []}
not_provided_sections = [k for k in SECTION_KEYS if sections.get(k) is None]
considered = [sections[k] for k in SECTION_KEYS if sections.get(k)]
failed = any((v.get("can_continue") is False) or v.get("status") == "failed" for v in considered)
warning = any(v.get("status") == "warning" for v in considered)
overall = "failed" if failed else ("warning" if warning else "passed")
action_items = []
if failed:
action_items.append("Investigate blocking checks before continuing.")
if warning:
action_items.append("Review warning checks and monitor downstream impact.")
if not_provided_sections:
action_items.append(f"Optional sections not provided: {', '.join(not_provided_sections)}.")
return {"run_id": runtime_context.get("run_id"), "dataset_name": runtime_context.get("dataset_name") or (contract or {}).get("dataset", {}).get("name"), "environment": runtime_context.get("environment"), "source_table": runtime_context.get("source_table"), "target_table": runtime_context.get("target_table"), "started_at_utc": runtime_context.get("started_at_utc"), "generated_at_utc": datetime.now(timezone.utc).isoformat(), "overall_status": overall, "can_continue": not failed, "sections": sections, "not_provided_sections": not_provided_sections, "action_items": action_items}