fix(ops): project stockplatform live freshness into priority readback

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -1036,6 +1036,30 @@ def apply_stockplatform_public_api_runtime_readback(
     state["stockplatform_public_api_data_dependency_classification"] = (
         data_dependency_classification
     )
+    state["stockplatform_freshness_status"] = str(
+        runtime_readback_body.get("freshness_status") or "unknown"
+    )
+    state["stockplatform_ingestion_status"] = str(
+        runtime_readback_body.get("ingestion_status") or "unknown"
+    )
+    state["stockplatform_latest_trading_date"] = str(
+        runtime_readback_body.get("freshness_latest_trading_date") or ""
+    )
+    state["stockplatform_freshness_blockers"] = _strings(
+        runtime_readback_body.get("freshness_blockers")
+    )
+    state["stockplatform_ingestion_blockers"] = _strings(
+        runtime_readback_body.get("ingestion_blockers")
+    )
+    state["stockplatform_freshness_sla_source_count"] = _int(
+        runtime_readback_body.get("freshness_sla_source_count")
+    )
+    state["stockplatform_freshness_source_count"] = _int(
+        runtime_readback_body.get("freshness_source_count")
+    )
+    state["stockplatform_freshness_non_ok_source_count"] = _int(
+        runtime_readback_body.get("freshness_non_ok_source_count")
+    )
     state["stockplatform_public_api_postgres_not_ready"] = postgres_not_ready
     state["stockplatform_postgres_readiness_or_data_source_readback_status"] = str(
         data_readiness_readback.get("status") or "unknown"
@@ -1078,6 +1102,28 @@ def apply_stockplatform_public_api_runtime_readback(
         evidence["stockplatform_data_dependency_classification"] = (
             data_dependency_classification
         )
+        evidence["stockplatform_freshness_status"] = state[
+            "stockplatform_freshness_status"
+        ]
+        evidence["stockplatform_ingestion_status"] = state[
+            "stockplatform_ingestion_status"
+        ]
+        evidence["stock_latest_trading_date"] = state[
+            "stockplatform_latest_trading_date"
+        ]
+        evidence["stock_blockers"] = state["stockplatform_freshness_blockers"]
+        evidence["stock_ingestion_blockers"] = state[
+            "stockplatform_ingestion_blockers"
+        ]
+        evidence["stockplatform_freshness_sla_source_count"] = state[
+            "stockplatform_freshness_sla_source_count"
+        ]
+        evidence["stockplatform_freshness_source_count"] = state[
+            "stockplatform_freshness_source_count"
+        ]
+        evidence["stockplatform_freshness_non_ok_source_count"] = state[
+            "stockplatform_freshness_non_ok_source_count"
+        ]
         evidence["stockplatform_postgres_not_ready"] = postgres_not_ready
         evidence[
             "stockplatform_postgres_readiness_or_data_source_readback_status"
@@ -3219,6 +3265,30 @@ def _refresh_rollups_after_stockplatform_overlay(
         state.get("stockplatform_postgres_readiness_or_data_source_postgres_ready")
         is True
     )
+    rollups["stockplatform_freshness_status"] = str(
+        state.get("stockplatform_freshness_status") or "unknown"
+    )
+    rollups["stockplatform_ingestion_status"] = str(
+        state.get("stockplatform_ingestion_status") or "unknown"
+    )
+    rollups["stockplatform_latest_trading_date"] = str(
+        state.get("stockplatform_latest_trading_date") or ""
+    )
+    rollups["stockplatform_freshness_blockers"] = _strings(
+        state.get("stockplatform_freshness_blockers")
+    )
+    rollups["stockplatform_ingestion_blockers"] = _strings(
+        state.get("stockplatform_ingestion_blockers")
+    )
+    rollups["stockplatform_freshness_sla_source_count"] = _int(
+        state.get("stockplatform_freshness_sla_source_count")
+    )
+    rollups["stockplatform_freshness_source_count"] = _int(
+        state.get("stockplatform_freshness_source_count")
+    )
+    rollups["stockplatform_freshness_non_ok_source_count"] = _int(
+        state.get("stockplatform_freshness_non_ok_source_count")
+    )
     rollups["active_p0_live_active_blocker_count"] = len(active_blockers)
     rollups["active_p0_event_gated_by_fresh_reboot_window_only"] = (
         rollups.get("active_p0_event_gated_by_fresh_reboot_window_only") is True
@@ -3251,6 +3321,30 @@ def _refresh_rollups_after_stockplatform_overlay(
         state.get("stockplatform_postgres_readiness_or_data_source_receipt_provided")
         is True
     )
+    summary["stockplatform_freshness_status"] = str(
+        state.get("stockplatform_freshness_status") or "unknown"
+    )
+    summary["stockplatform_ingestion_status"] = str(
+        state.get("stockplatform_ingestion_status") or "unknown"
+    )
+    summary["stockplatform_latest_trading_date"] = str(
+        state.get("stockplatform_latest_trading_date") or ""
+    )
+    summary["stockplatform_freshness_blockers"] = _strings(
+        state.get("stockplatform_freshness_blockers")
+    )
+    summary["stockplatform_ingestion_blockers"] = _strings(
+        state.get("stockplatform_ingestion_blockers")
+    )
+    summary["stockplatform_freshness_sla_source_count"] = _int(
+        state.get("stockplatform_freshness_sla_source_count")
+    )
+    summary["stockplatform_freshness_source_count"] = _int(
+        state.get("stockplatform_freshness_source_count")
+    )
+    summary["stockplatform_freshness_non_ok_source_count"] = _int(
+        state.get("stockplatform_freshness_non_ok_source_count")
+    )
     summary["next_executable_mainline_workplan_id"] = str(
         state.get("next_executable_mainline_workplan_id") or ""
     )

apps/api/src/services/stockplatform_public_api_runtime_readback.py

@@ -114,6 +114,7 @@ def _build_payload(
     ingestion = _dict(probes.get("ingestion"))
     freshness_json = _dict(freshness.get("json"))
     ingestion_json = _dict(ingestion.get("json"))
+    freshness_sources = _sources(freshness_json.get("sources"))
 
     checks = {
         "public_web_health_ok": web.get("http_status") == 200,
@@ -196,8 +197,19 @@ def _build_payload(
             "ingestion_http_status": ingestion.get("http_status"),
             "freshness_status": str(freshness_json.get("status") or "unknown"),
             "ingestion_status": str(ingestion_json.get("status") or "unknown"),
+            "freshness_latest_trading_date": str(
+                freshness_json.get("latest_trading_date") or ""
+            ),
+            "ingestion_latest_trading_date": str(
+                ingestion_json.get("latest_trading_date") or ""
+            ),
             "freshness_blockers": _strings(freshness_json.get("blockers")),
             "ingestion_blockers": _strings(ingestion_json.get("blockers")),
+            "freshness_source_count": len(freshness_sources),
+            "freshness_sla_source_count": _sla_source_count(freshness_sources),
+            "freshness_non_ok_source_count": _non_ok_source_count(
+                freshness_sources
+            ),
             "data_dependency_classification": data_dependency_classification,
             "postgres_not_ready": postgres_not_ready,
             "committed_freshness_status": str(
@@ -408,6 +420,7 @@ def _data_readiness_control_readback(
         missing_evidence.append("freshness_endpoint_http_200")
     if ingestion_http_status != 200:
         missing_evidence.append("ingestion_endpoint_http_200")
+    freshness_sources = _sources(freshness_json.get("sources"))
     return {
         "schema_version": _DATA_READINESS_SCHEMA_VERSION,
         "receipt_key": _DATA_READINESS_RECEIPT_KEY,
@@ -424,8 +437,17 @@ def _data_readiness_control_readback(
         "ingestion_http_status": ingestion_http_status,
         "freshness_status": str(freshness_json.get("status") or "unknown"),
         "ingestion_status": str(ingestion_json.get("status") or "unknown"),
+        "freshness_latest_trading_date": str(
+            freshness_json.get("latest_trading_date") or ""
+        ),
+        "ingestion_latest_trading_date": str(
+            ingestion_json.get("latest_trading_date") or ""
+        ),
         "freshness_blockers": _strings(freshness_json.get("blockers")),
         "ingestion_blockers": _strings(ingestion_json.get("blockers")),
+        "freshness_source_count": len(freshness_sources),
+        "freshness_sla_source_count": _sla_source_count(freshness_sources),
+        "freshness_non_ok_source_count": _non_ok_source_count(freshness_sources),
         "active_blockers": [
             blocker
             for blocker in data_dependency_blockers
@@ -648,6 +670,20 @@ def _dict(value: Any) -> dict[str, Any]:
     return value if isinstance(value, dict) else {}
 
 
+def _sources(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list):
+        return []
+    return [item for item in value if isinstance(item, dict)]
+
+
+def _sla_source_count(sources: list[dict[str, Any]]) -> int:
+    return sum(1 for source in sources if isinstance(source.get("sla"), dict))
+
+
+def _non_ok_source_count(sources: list[dict[str, Any]]) -> int:
+    return sum(1 for source in sources if source.get("status") != "ok")
+
+
 def _int_or_none(value: Any) -> int | None:
     if isinstance(value, bool):
         return int(value)