feat(stockplatform): surface data readiness receipt

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -994,6 +994,9 @@ def apply_stockplatform_public_api_runtime_readback(
     runtime_blockers = _strings(runtime_readback.get("active_blockers"))
     runtime_rollups = _dict(runtime_readback.get("rollups"))
     runtime_readback_body = _dict(runtime_readback.get("readback"))
+    data_readiness_readback = _dict(
+        runtime_readback.get("data_readiness_control_readback")
+    )
     data_dependency_classification = str(
         runtime_rollups.get("data_dependency_classification")
         or runtime_readback_body.get("data_dependency_classification")
@@ -1033,6 +1036,18 @@ def apply_stockplatform_public_api_runtime_readback(
         data_dependency_classification
     )
     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"
+    )
+    state["stockplatform_postgres_readiness_or_data_source_readback_present"] = bool(
+        data_readiness_readback.get("readback_present") is True
+    )
+    state["stockplatform_postgres_readiness_or_data_source_receipt_provided"] = bool(
+        data_readiness_readback.get("receipt_provided") is True
+    )
+    state["stockplatform_postgres_readiness_or_data_source_postgres_ready"] = bool(
+        data_readiness_readback.get("postgres_ready") is True
+    )
 
     for item in _list(payload.get("in_progress_or_blocked_in_priority_order")):
         workplan = _dict(item)
@@ -1063,6 +1078,18 @@ def apply_stockplatform_public_api_runtime_readback(
             data_dependency_classification
         )
         evidence["stockplatform_postgres_not_ready"] = postgres_not_ready
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_readback_status"
+        ] = state["stockplatform_postgres_readiness_or_data_source_readback_status"]
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_readback_present"
+        ] = state["stockplatform_postgres_readiness_or_data_source_readback_present"]
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_receipt_provided"
+        ] = state["stockplatform_postgres_readiness_or_data_source_receipt_provided"]
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_postgres_ready"
+        ] = state["stockplatform_postgres_readiness_or_data_source_postgres_ready"]
         evidence["stockplatform_recovery_control_path_status"] = (
             recovery_control_path_status
         )
@@ -1079,12 +1106,14 @@ def apply_stockplatform_public_api_runtime_readback(
         )
         workplan["reason"] = (
             _stockplatform_runtime_drift_reason(
+                data_dependency_classification=data_dependency_classification,
                 postgres_not_ready=postgres_not_ready,
             )
         )
         professional_fix = _dict(workplan.setdefault("professional_fix", {}))
         professional_fix["action"] = (
             _stockplatform_runtime_drift_action(
+                data_dependency_classification=data_dependency_classification,
                 postgres_not_ready=postgres_not_ready,
             )
         )
@@ -1108,11 +1137,15 @@ def apply_stockplatform_public_api_runtime_readback(
         "P0-006-STOCKPLATFORM-PUBLIC-API-RUNTIME-READBACK"
     )
     state["next_executable_mainline_state"] = (
-        _stockplatform_runtime_next_state(postgres_not_ready=postgres_not_ready)
+        _stockplatform_runtime_next_state(
+            data_dependency_classification=data_dependency_classification,
+            postgres_not_ready=postgres_not_ready,
+        )
     )
     payload["status"] = "p0_006_blocked_stockplatform_public_api_runtime_drift"
     payload["next_execution_order"] = [
         _stockplatform_runtime_first_order(
+            data_dependency_classification=data_dependency_classification,
             postgres_not_ready=postgres_not_ready,
         ),
         (
@@ -1152,6 +1185,9 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
     source_diff = _dict(recovery_preflight.get("source_of_truth_diff"))
     target_selector = _dict(recovery_preflight.get("target_selector"))
     controlled_policy = _dict(recovery_preflight.get("controlled_apply_policy"))
+    data_readiness_readback = _dict(
+        recovery_preflight.get("data_readiness_control_readback")
+    )
     candidate_packaged = bool(
         rollups.get("controlled_recovery_candidate_packaged") is True
     )
@@ -1167,6 +1203,18 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
     state["stockplatform_public_api_controlled_recovery_data_dependency"] = (
         data_dependency_classification
     )
+    state["stockplatform_postgres_readiness_or_data_source_readback_status"] = str(
+        data_readiness_readback.get("status") or "unknown"
+    )
+    state["stockplatform_postgres_readiness_or_data_source_readback_present"] = bool(
+        data_readiness_readback.get("readback_present") is True
+    )
+    state["stockplatform_postgres_readiness_or_data_source_receipt_provided"] = bool(
+        data_readiness_readback.get("receipt_provided") is True
+    )
+    state["stockplatform_postgres_readiness_or_data_source_postgres_ready"] = bool(
+        data_readiness_readback.get("postgres_ready") is True
+    )
     state["stockplatform_public_api_controlled_recovery_active_blockers"] = _strings(
         recovery_preflight.get("active_blockers")
     )
@@ -1208,6 +1256,18 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
         evidence["stockplatform_source_of_truth_diff_status"] = str(
             source_diff.get("status") or ""
         )
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_readback_status"
+        ] = state["stockplatform_postgres_readiness_or_data_source_readback_status"]
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_readback_present"
+        ] = state["stockplatform_postgres_readiness_or_data_source_readback_present"]
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_receipt_provided"
+        ] = state["stockplatform_postgres_readiness_or_data_source_receipt_provided"]
+        evidence[
+            "stockplatform_postgres_readiness_or_data_source_postgres_ready"
+        ] = state["stockplatform_postgres_readiness_or_data_source_postgres_ready"]
         evidence["stockplatform_controlled_recovery_candidate_packaged"] = (
             candidate_packaged
         )
@@ -1291,6 +1351,14 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
     summary["stockplatform_public_api_host_control_path_readback_required"] = bool(
         rollups.get("host_control_path_readback_required") is True
     )
+    summary["stockplatform_postgres_readiness_or_data_source_readback_status"] = str(
+        state.get("stockplatform_postgres_readiness_or_data_source_readback_status")
+        or "unknown"
+    )
+    summary["stockplatform_postgres_readiness_or_data_source_receipt_provided"] = bool(
+        state.get("stockplatform_postgres_readiness_or_data_source_receipt_provided")
+        is True
+    )
 
 
 def apply_harbor_registry_controlled_recovery_preflight(
@@ -2493,19 +2561,32 @@ def _stockplatform_recovery_first_order(
     )
 
 
-def _stockplatform_runtime_next_state(*, postgres_not_ready: bool) -> str:
+def _stockplatform_runtime_next_state(
+    *,
+    data_dependency_classification: str,
+    postgres_not_ready: bool,
+) -> str:
     if postgres_not_ready:
         return (
             "blocked_live_stockplatform_public_api_requires_production_"
             "migration_control_channel_readback"
         )
+    if data_dependency_classification == "freshness_or_ingestion_not_ready":
+        return (
+            "blocked_live_stockplatform_public_api_requires_data_dependency_"
+            "source_contract_readback"
+        )
     return (
         "blocked_live_stockplatform_public_api_requires_separate_runtime_"
         "control_path_recovery_without_daemon_restart"
     )
 
 
-def _stockplatform_runtime_drift_reason(*, postgres_not_ready: bool) -> str:
+def _stockplatform_runtime_drift_reason(
+    *,
+    data_dependency_classification: str,
+    postgres_not_ready: bool,
+) -> str:
     if postgres_not_ready:
         return (
             "StockPlatform public API health is now HTTP 200, but freshness and "
@@ -2513,6 +2594,13 @@ def _stockplatform_runtime_drift_reason(*, postgres_not_ready: bool) -> str:
             "the production migration/control-channel path before claiming the "
             "reboot SLO lane is only waiting on a fresh boot window."
         )
+    if data_dependency_classification == "freshness_or_ingestion_not_ready":
+        return (
+            "StockPlatform public API health is now HTTP 200, but freshness and "
+            "ingestion readbacks are not green. P0-006 must stay on the data "
+            "dependency/source contract readback path before claiming the reboot "
+            "SLO lane is only waiting on a fresh boot window."
+        )
     return (
         "Committed P0-006 scorecard says StockPlatform freshness and ingestion "
         "are ok, but live public API readback is blocked. This must remain "
@@ -2521,7 +2609,11 @@ def _stockplatform_runtime_drift_reason(*, postgres_not_ready: bool) -> str:
     )
 
 
-def _stockplatform_runtime_drift_action(*, postgres_not_ready: bool) -> str:
+def _stockplatform_runtime_drift_action(
+    *,
+    data_dependency_classification: str,
+    postgres_not_ready: bool,
+) -> str:
     if postgres_not_ready:
         return (
             "Recover the StockPlatform production migration/control channel "
@@ -2531,6 +2623,15 @@ def _stockplatform_runtime_drift_action(*, postgres_not_ready: bool) -> str:
             "reload Nginx, manually write DB rows, fake freshness, trigger "
             "unrelated workflows, or read secrets from this lane."
         )
+    if data_dependency_classification == "freshness_or_ingestion_not_ready":
+        return (
+            "Run the StockPlatform data dependency/source contract readback "
+            "with target selector, source-of-truth diff, dry-run/check-mode, "
+            "rollback, and public API verifier, then rerun the public readback. "
+            "Do not restart Docker daemon, reboot hosts, reload Nginx, manually "
+            "write DB rows, fake freshness, trigger unrelated workflows, or read "
+            "secrets from this lane."
+        )
     return (
         "Recover the StockPlatform API container/runtime control path with a "
         "separate bounded plan, then rerun public API readback. Do not restart "
@@ -2539,7 +2640,11 @@ def _stockplatform_runtime_drift_action(*, postgres_not_ready: bool) -> str:
     )
 
 
-def _stockplatform_runtime_first_order(*, postgres_not_ready: bool) -> str:
+def _stockplatform_runtime_first_order(
+    *,
+    data_dependency_classification: str,
+    postgres_not_ready: bool,
+) -> str:
     if postgres_not_ready:
         return (
             "P0-006-STOCKPLATFORM-PUBLIC-API-RUNTIME-READBACK: live public "
@@ -2547,6 +2652,13 @@ def _stockplatform_runtime_first_order(*, postgres_not_ready: bool) -> str:
             "postgres_not_ready; run the production migration/control-channel "
             "path before claiming the lane is only reboot-window gated."
         )
+    if data_dependency_classification == "freshness_or_ingestion_not_ready":
+        return (
+            "P0-006-STOCKPLATFORM-PUBLIC-API-RUNTIME-READBACK: live public "
+            "StockPlatform API health is 200 but freshness/ingestion are not "
+            "green; run the data dependency/source contract readback before "
+            "claiming the lane is only reboot-window gated."
+        )
     return (
         "P0-006-STOCKPLATFORM-PUBLIC-API-RUNTIME-READBACK: live public "
         "StockPlatform API is blocked while committed scorecard still says "
@@ -3094,6 +3206,18 @@ def _refresh_rollups_after_stockplatform_overlay(
     rollups["stockplatform_public_api_recovery_control_path_blocker_count"] = len(
         recovery_control_path_blockers
     )
+    rollups["stockplatform_postgres_readiness_or_data_source_readback_present"] = bool(
+        state.get("stockplatform_postgres_readiness_or_data_source_readback_present")
+        is True
+    )
+    rollups["stockplatform_postgres_readiness_or_data_source_receipt_provided"] = bool(
+        state.get("stockplatform_postgres_readiness_or_data_source_receipt_provided")
+        is True
+    )
+    rollups["stockplatform_postgres_readiness_or_data_source_postgres_ready"] = bool(
+        state.get("stockplatform_postgres_readiness_or_data_source_postgres_ready")
+        is True
+    )
     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
@@ -3118,6 +3242,14 @@ def _refresh_rollups_after_stockplatform_overlay(
     summary["stockplatform_public_api_recovery_control_path_blockers"] = (
         recovery_control_path_blockers
     )
+    summary["stockplatform_postgres_readiness_or_data_source_readback_status"] = str(
+        state.get("stockplatform_postgres_readiness_or_data_source_readback_status")
+        or "unknown"
+    )
+    summary["stockplatform_postgres_readiness_or_data_source_receipt_provided"] = bool(
+        state.get("stockplatform_postgres_readiness_or_data_source_receipt_provided")
+        is True
+    )
     summary["next_executable_mainline_workplan_id"] = str(
         state.get("next_executable_mainline_workplan_id") or ""
     )

apps/api/src/services/stockplatform_public_api_controlled_recovery_preflight.py

@@ -48,6 +48,9 @@ def _build_payload(
     runtime_ready = runtime.get("status") == "stockplatform_public_api_runtime_ready"
     readback = _dict(runtime.get("readback"))
     rollups = _dict(runtime.get("rollups"))
+    data_readiness_control_readback = _dict(
+        runtime.get("data_readiness_control_readback")
+    )
     api_layer_classification = _classify_api_layer(readback, rollups, runtime_ready)
     data_dependency_classification = _data_dependency_classification(runtime)
     route_target_ready = route_contract.get("route_target_ready") is True
@@ -84,6 +87,7 @@ def _build_payload(
             candidate_packaged=candidate_packaged,
             route_target_ready=route_target_ready,
             api_layer_classification=api_layer_classification,
+            data_dependency_classification=data_dependency_classification,
         ),
         "active_blockers": active_blockers,
         "active_blocker_count": len(active_blockers),
@@ -118,9 +122,16 @@ def _build_payload(
             "ingestion_http_status": readback.get("ingestion_http_status"),
             "http_502_count": _int(rollups.get("http_502_count")),
             "data_dependency_classification": data_dependency_classification,
+            "data_readiness_control_readback_status": (
+                data_readiness_control_readback.get("status")
+            ),
+            "data_readiness_control_receipt_provided": (
+                data_readiness_control_readback.get("receipt_provided") is True
+            ),
             "route_source_file": str(route_contract.get("route_source_file") or ""),
             "route_target": str(route_contract.get("route_target") or ""),
         },
+        "data_readiness_control_readback": data_readiness_control_readback,
         "controlled_apply_policy": {
             "risk_level": "high",
             "break_glass_required": False,
@@ -184,6 +195,12 @@ def _build_payload(
                 _post_apply_verifiers(runtime_ready=runtime_ready)
             ),
             "learning_writeback_contract_count": len(_learning_writeback_contracts()),
+            "postgres_readiness_or_data_source_readback_present": (
+                data_readiness_control_readback.get("readback_present") is True
+            ),
+            "postgres_readiness_or_data_source_receipt_provided": (
+                data_readiness_control_readback.get("receipt_provided") is True
+            ),
         },
         "operation_boundaries": {
             "read_only_public_https_probe": True,
@@ -258,12 +275,20 @@ def _classify_api_layer(
 
 
 def _data_dependency_classification(runtime: dict[str, Any]) -> str:
+    runtime_rollups = _dict(runtime.get("rollups"))
+    runtime_readback = _dict(runtime.get("readback"))
+    live_classification = str(
+        runtime_rollups.get("data_dependency_classification")
+        or runtime_readback.get("data_dependency_classification")
+        or ""
+    )
+    if live_classification and live_classification != "none":
+        return live_classification
     blockers = _strings(runtime.get("active_blockers"))
-    readback = _dict(runtime.get("readback"))
-    if readback.get("api_health_http_status") != 200:
+    if runtime_readback.get("api_health_http_status") != 200:
         return "none"
     if (
-        readback.get("postgres_not_ready") is True
+        runtime_readback.get("postgres_not_ready") is True
         or any("postgres_not_ready" in blocker for blocker in blockers)
     ):
         return "postgres_not_ready"
@@ -395,6 +420,7 @@ def _safe_next_step(
     candidate_packaged: bool,
     route_target_ready: bool,
     api_layer_classification: str,
+    data_dependency_classification: str,
 ) -> str:
     if runtime_ready:
         return "stockplatform_public_api_ready_keep_p0_006_reboot_window_gate"
@@ -402,6 +428,11 @@ def _safe_next_step(
         candidate_packaged
         and api_layer_classification == "api_live_data_dependency_not_ready"
     ):
+        if data_dependency_classification != "postgres_not_ready":
+            return (
+                "run_stockplatform_data_dependency_source_contract_readback_"
+                "then_public_api_verifier"
+            )
         return (
             "run_stockplatform_production_migration_path_after_control_channel_"
             "readback_then_public_api_verifier"

apps/api/src/services/stockplatform_public_api_runtime_readback.py

@@ -19,6 +19,12 @@ from src.services.reboot_auto_recovery_slo_scorecard import (
 from src.services.snapshot_paths import default_operations_dir
 
 _API_SCHEMA_VERSION = "stockplatform_public_api_runtime_readback_v1"
+_DATA_READINESS_SCHEMA_VERSION = (
+    "stockplatform_postgres_readiness_or_data_source_readback_v1"
+)
+_DATA_READINESS_RECEIPT_KEY = (
+    "stockplatform_postgres_readiness_or_data_source_readback"
+)
 _DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
 _RECOVERY_RECEIPT_FILE = (
     "stockplatform-public-api-runtime-recovery-control-receipt.snapshot.json"
@@ -125,19 +131,30 @@ def _build_payload(
         ingestion_json=ingestion_json,
     )
     ready = all(checks.values())
-    recovery_control_path = _recovery_control_path_readback(
-        runtime_ready=ready,
-        receipt=recovery_control_receipt,
-    )
-    recovery_control_path_blockers = _strings(
-        recovery_control_path.get("active_blockers")
-    )
-    active_blockers = _unique_strings(active_blockers + recovery_control_path_blockers)
     data_dependency_classification = _data_dependency_classification(
         active_blockers,
         api_health_ok=checks["public_api_health_ok"],
     )
     postgres_not_ready = data_dependency_classification == "postgres_not_ready"
+    data_readiness_control_readback = _data_readiness_control_readback(
+        checks=checks,
+        freshness=freshness,
+        ingestion=ingestion,
+        freshness_json=freshness_json,
+        ingestion_json=ingestion_json,
+        data_dependency_classification=data_dependency_classification,
+        postgres_not_ready=postgres_not_ready,
+        data_dependency_blockers=active_blockers,
+    )
+    recovery_control_path = _recovery_control_path_readback(
+        runtime_ready=ready,
+        receipt=recovery_control_receipt,
+        data_readiness_control_readback=data_readiness_control_readback,
+    )
+    recovery_control_path_blockers = _strings(
+        recovery_control_path.get("active_blockers")
+    )
+    active_blockers = _unique_strings(active_blockers + recovery_control_path_blockers)
     committed_freshness_ok = committed_stockplatform.get("freshness_status") == "ok"
     committed_ingestion_ok = committed_stockplatform.get("ingestion_status") == "ok"
     committed_ok = committed_freshness_ok and committed_ingestion_ok
@@ -151,7 +168,7 @@ def _build_payload(
         "stockplatform_public_api_ready_keep_p0_006_reboot_window_gate"
         if ready
         else _blocked_safe_next_step(
-            postgres_not_ready=postgres_not_ready,
+            data_dependency_classification=data_dependency_classification,
         )
     )
     http_statuses = {
@@ -194,6 +211,12 @@ def _build_payload(
             ),
             "recovery_control_path_status": recovery_control_path.get("status"),
             "recovery_control_path_active_blockers": recovery_control_path_blockers,
+            "data_readiness_control_readback_status": (
+                data_readiness_control_readback.get("status")
+            ),
+            "data_readiness_control_readback_present": (
+                data_readiness_control_readback.get("readback_present") is True
+            ),
         },
         "rollups": {
             "runtime_ready": ready,
@@ -224,7 +247,14 @@ def _build_payload(
             ),
             "data_dependency_classification": data_dependency_classification,
             "postgres_not_ready": postgres_not_ready,
+            "postgres_readiness_or_data_source_readback_present": (
+                data_readiness_control_readback.get("readback_present") is True
+            ),
+            "postgres_readiness_or_data_source_receipt_provided": (
+                data_readiness_control_readback.get("receipt_provided") is True
+            ),
         },
+        "data_readiness_control_readback": data_readiness_control_readback,
         "recovery_control_path": recovery_control_path,
         "probes": probes,
         "operation_boundaries": {
@@ -249,20 +279,38 @@ def _recovery_control_path_readback(
     *,
     runtime_ready: bool,
     receipt: dict[str, Any],
+    data_readiness_control_readback: dict[str, Any],
 ) -> dict[str, Any]:
     if runtime_ready:
         return {
             "status": "not_required_stockplatform_runtime_ready",
             "active_blockers": [],
             "receipt": receipt,
+            "provided_receipts": _provided_receipts_with_live_readback(
+                _strings(receipt.get("provided_receipts")),
+                data_readiness_control_readback,
+            ),
+            "missing_receipts": _missing_receipts_after_live_readback(
+                _strings(receipt.get("missing_receipts")),
+                data_readiness_control_readback,
+            ),
             "safe_recovery_channels": [
                 "keep_monitoring_public_api_runtime_readback",
             ],
         }
     if receipt:
-        active_blockers = _unique_strings(
-            _strings(receipt.get("active_blockers"))
-            or _strings(receipt.get("missing_receipts"))
+        provided_receipts = _provided_receipts_with_live_readback(
+            _strings(receipt.get("provided_receipts")),
+            data_readiness_control_readback,
+        )
+        missing_receipts = _missing_receipts_after_live_readback(
+            _strings(receipt.get("missing_receipts")),
+            data_readiness_control_readback,
+        )
+        active_blockers = _recovery_control_active_blockers(
+            receipt_blockers=_strings(receipt.get("active_blockers")),
+            missing_receipts=missing_receipts,
+            data_readiness_control_readback=data_readiness_control_readback,
         )
         return {
             "status": str(
@@ -271,8 +319,13 @@ def _recovery_control_path_readback(
             ),
             "active_blockers": active_blockers,
             "receipt": receipt,
-            "provided_receipts": _strings(receipt.get("provided_receipts")),
-            "missing_receipts": _strings(receipt.get("missing_receipts")),
+            "provided_receipts": provided_receipts,
+            "missing_receipts": missing_receipts,
+            "live_data_readiness_receipt_key": (
+                _DATA_READINESS_RECEIPT_KEY
+                if data_readiness_control_readback.get("receipt_provided") is True
+                else ""
+            ),
             "safe_recovery_channels": _strings(
                 receipt.get("safe_recovery_channels")
             ),
@@ -310,6 +363,130 @@ def _recovery_control_path_readback(
     }
 
 
+def _data_readiness_control_readback(
+    *,
+    checks: dict[str, bool],
+    freshness: dict[str, Any],
+    ingestion: dict[str, Any],
+    freshness_json: dict[str, Any],
+    ingestion_json: dict[str, Any],
+    data_dependency_classification: str,
+    postgres_not_ready: bool,
+    data_dependency_blockers: list[str],
+) -> dict[str, Any]:
+    freshness_http_status = freshness.get("http_status")
+    ingestion_http_status = ingestion.get("http_status")
+    readback_present = freshness_http_status == 200 and ingestion_http_status == 200
+    data_ready = (
+        readback_present
+        and freshness_json.get("status") == "ok"
+        and ingestion_json.get("status") == "ok"
+    )
+    if data_ready:
+        status = "stockplatform_data_readiness_ready"
+        safe_next_step = "run_post_apply_public_api_verifier_and_clear_p0_006_blocker"
+    elif readback_present and postgres_not_ready:
+        status = "blocked_stockplatform_postgres_not_ready"
+        safe_next_step = (
+            "run_stockplatform_production_migration_path_after_control_channel_"
+            "readback_no_manual_db_write"
+        )
+    elif readback_present:
+        status = "blocked_stockplatform_data_dependency_not_ready"
+        safe_next_step = (
+            "diff_stockplatform_freshness_ingestion_source_contract_then_rerun_"
+            "public_api_verifier"
+        )
+    else:
+        status = "blocked_stockplatform_data_readiness_readback_unavailable"
+        safe_next_step = (
+            "restore_public_api_data_endpoints_or_control_channel_readback_then_"
+            "rerun_data_readiness_readback"
+        )
+    missing_evidence = []
+    if freshness_http_status != 200:
+        missing_evidence.append("freshness_endpoint_http_200")
+    if ingestion_http_status != 200:
+        missing_evidence.append("ingestion_endpoint_http_200")
+    return {
+        "schema_version": _DATA_READINESS_SCHEMA_VERSION,
+        "receipt_key": _DATA_READINESS_RECEIPT_KEY,
+        "status": status,
+        "source": "stockplatform_public_api_freshness_ingestion_probe",
+        "readback_present": readback_present,
+        "receipt_provided": readback_present,
+        "safe_next_step": safe_next_step,
+        "data_ready": data_ready,
+        "postgres_ready": data_ready and not postgres_not_ready,
+        "postgres_not_ready": postgres_not_ready,
+        "data_dependency_classification": data_dependency_classification,
+        "freshness_http_status": freshness_http_status,
+        "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_blockers": _strings(freshness_json.get("blockers")),
+        "ingestion_blockers": _strings(ingestion_json.get("blockers")),
+        "active_blockers": [
+            blocker
+            for blocker in data_dependency_blockers
+            if blocker.startswith("stockplatform_freshness_")
+            or blocker.startswith("stockplatform_ingestion_")
+        ],
+        "missing_evidence": missing_evidence,
+        "operation_boundaries": {
+            "public_https_probe_only": True,
+            "ssh_used": False,
+            "docker_command_performed": False,
+            "database_write_or_restore_performed": False,
+            "stockplatform_manual_data_write_performed": False,
+            "secret_value_collection_allowed": False,
+            "runtime_action_performed": False,
+        },
+    }
+
+
+def _provided_receipts_with_live_readback(
+    provided_receipts: list[str],
+    data_readiness_control_readback: dict[str, Any],
+) -> list[str]:
+    if data_readiness_control_readback.get("receipt_provided") is not True:
+        return provided_receipts
+    return _unique_strings([*provided_receipts, _DATA_READINESS_RECEIPT_KEY])
+
+
+def _missing_receipts_after_live_readback(
+    missing_receipts: list[str],
+    data_readiness_control_readback: dict[str, Any],
+) -> list[str]:
+    if data_readiness_control_readback.get("receipt_provided") is not True:
+        return missing_receipts
+    return [
+        receipt
+        for receipt in missing_receipts
+        if receipt != _DATA_READINESS_RECEIPT_KEY
+    ]
+
+
+def _recovery_control_active_blockers(
+    *,
+    receipt_blockers: list[str],
+    missing_receipts: list[str],
+    data_readiness_control_readback: dict[str, Any],
+) -> list[str]:
+    blockers = receipt_blockers or missing_receipts
+    if data_readiness_control_readback.get("receipt_provided") is not True:
+        return _unique_strings(blockers)
+    if data_readiness_control_readback.get("postgres_not_ready") is True:
+        return _unique_strings(blockers)
+    return _unique_strings(
+        [
+            blocker
+            for blocker in blockers
+            if blocker != "stockplatform_postgres_not_ready"
+        ]
+    )
+
+
 def _load_recovery_control_receipt(operations_dir: Path) -> dict[str, Any]:
     path = operations_dir / _RECOVERY_RECEIPT_FILE
     if not path.exists():
@@ -327,12 +504,17 @@ def _load_recovery_control_receipt(operations_dir: Path) -> dict[str, Any]:
     return payload
 
 
-def _blocked_safe_next_step(*, postgres_not_ready: bool) -> str:
-    if postgres_not_ready:
+def _blocked_safe_next_step(*, data_dependency_classification: str) -> str:
+    if data_dependency_classification == "postgres_not_ready":
         return (
             "run_stockplatform_production_migration_path_after_control_channel_"
             "readback_then_rerun_public_api_readback"
         )
+    if data_dependency_classification == "freshness_or_ingestion_not_ready":
+        return (
+            "run_stockplatform_data_dependency_source_contract_readback_then_"
+            "rerun_public_api_verifier"
+        )
     return (
         "recover_stockplatform_api_container_runtime_or_docker_control_path_"
         "without_daemon_restart_then_rerun_public_api_readback"