feat(stockplatform): surface source contract receipt

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -998,6 +998,9 @@ def apply_stockplatform_public_api_runtime_readback(
     data_readiness_readback = _dict(
         runtime_readback.get("data_readiness_control_readback")
     )
+    source_contract_readback = _dict(
+        runtime_readback.get("source_contract_control_readback")
+    )
     data_dependency_classification = str(
         runtime_rollups.get("data_dependency_classification")
         or runtime_readback_body.get("data_dependency_classification")
@@ -1073,6 +1076,24 @@ def apply_stockplatform_public_api_runtime_readback(
     state["stockplatform_postgres_readiness_or_data_source_postgres_ready"] = bool(
         data_readiness_readback.get("postgres_ready") is True
     )
+    state["stockplatform_source_contract_readback_status"] = str(
+        source_contract_readback.get("status") or "unknown"
+    )
+    state["stockplatform_source_contract_readback_present"] = bool(
+        source_contract_readback.get("readback_present") is True
+    )
+    state["stockplatform_source_contract_receipt_provided"] = bool(
+        source_contract_readback.get("receipt_provided") is True
+    )
+    state["stockplatform_source_contract_ready"] = bool(
+        source_contract_readback.get("source_contract_ready") is True
+    )
+    state["stockplatform_source_contract_non_ok_source_count"] = _int(
+        source_contract_readback.get("non_ok_source_count")
+    )
+    state["stockplatform_source_contract_blocking_sources"] = _list(
+        source_contract_readback.get("blocking_sources")
+    )
 
     for item in _list(payload.get("in_progress_or_blocked_in_priority_order")):
         workplan = _dict(item)
@@ -1137,6 +1158,24 @@ def apply_stockplatform_public_api_runtime_readback(
         evidence[
             "stockplatform_postgres_readiness_or_data_source_postgres_ready"
         ] = state["stockplatform_postgres_readiness_or_data_source_postgres_ready"]
+        evidence["stockplatform_source_contract_readback_status"] = state[
+            "stockplatform_source_contract_readback_status"
+        ]
+        evidence["stockplatform_source_contract_readback_present"] = state[
+            "stockplatform_source_contract_readback_present"
+        ]
+        evidence["stockplatform_source_contract_receipt_provided"] = state[
+            "stockplatform_source_contract_receipt_provided"
+        ]
+        evidence["stockplatform_source_contract_ready"] = state[
+            "stockplatform_source_contract_ready"
+        ]
+        evidence["stockplatform_source_contract_non_ok_source_count"] = state[
+            "stockplatform_source_contract_non_ok_source_count"
+        ]
+        evidence["stockplatform_source_contract_blocking_sources"] = state[
+            "stockplatform_source_contract_blocking_sources"
+        ]
         evidence["stockplatform_recovery_control_path_status"] = (
             recovery_control_path_status
         )
@@ -1235,6 +1274,9 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
     data_readiness_readback = _dict(
         recovery_preflight.get("data_readiness_control_readback")
     )
+    source_contract_readback = _dict(
+        recovery_preflight.get("source_contract_control_readback")
+    )
     candidate_packaged = bool(
         rollups.get("controlled_recovery_candidate_packaged") is True
     )
@@ -1262,6 +1304,24 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
     state["stockplatform_postgres_readiness_or_data_source_postgres_ready"] = bool(
         data_readiness_readback.get("postgres_ready") is True
     )
+    state["stockplatform_source_contract_readback_status"] = str(
+        source_contract_readback.get("status") or "unknown"
+    )
+    state["stockplatform_source_contract_readback_present"] = bool(
+        source_contract_readback.get("readback_present") is True
+    )
+    state["stockplatform_source_contract_receipt_provided"] = bool(
+        source_contract_readback.get("receipt_provided") is True
+    )
+    state["stockplatform_source_contract_ready"] = bool(
+        source_contract_readback.get("source_contract_ready") is True
+    )
+    state["stockplatform_source_contract_non_ok_source_count"] = _int(
+        source_contract_readback.get("non_ok_source_count")
+    )
+    state["stockplatform_source_contract_blocking_sources"] = _list(
+        source_contract_readback.get("blocking_sources")
+    )
     state["stockplatform_public_api_controlled_recovery_active_blockers"] = _strings(
         recovery_preflight.get("active_blockers")
     )
@@ -1315,6 +1375,24 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
         evidence[
             "stockplatform_postgres_readiness_or_data_source_postgres_ready"
         ] = state["stockplatform_postgres_readiness_or_data_source_postgres_ready"]
+        evidence["stockplatform_source_contract_readback_status"] = state[
+            "stockplatform_source_contract_readback_status"
+        ]
+        evidence["stockplatform_source_contract_readback_present"] = state[
+            "stockplatform_source_contract_readback_present"
+        ]
+        evidence["stockplatform_source_contract_receipt_provided"] = state[
+            "stockplatform_source_contract_receipt_provided"
+        ]
+        evidence["stockplatform_source_contract_ready"] = state[
+            "stockplatform_source_contract_ready"
+        ]
+        evidence["stockplatform_source_contract_non_ok_source_count"] = state[
+            "stockplatform_source_contract_non_ok_source_count"
+        ]
+        evidence["stockplatform_source_contract_blocking_sources"] = state[
+            "stockplatform_source_contract_blocking_sources"
+        ]
         evidence["stockplatform_controlled_recovery_candidate_packaged"] = (
             candidate_packaged
         )
@@ -1406,6 +1484,15 @@ def apply_stockplatform_public_api_controlled_recovery_preflight(
         state.get("stockplatform_postgres_readiness_or_data_source_receipt_provided")
         is True
     )
+    summary["stockplatform_source_contract_readback_status"] = str(
+        state.get("stockplatform_source_contract_readback_status") or "unknown"
+    )
+    summary["stockplatform_source_contract_receipt_provided"] = bool(
+        state.get("stockplatform_source_contract_receipt_provided") is True
+    )
+    summary["stockplatform_source_contract_non_ok_source_count"] = _int(
+        state.get("stockplatform_source_contract_non_ok_source_count")
+    )
 
 
 def apply_harbor_registry_controlled_recovery_preflight(
@@ -3265,6 +3352,18 @@ def _refresh_rollups_after_stockplatform_overlay(
         state.get("stockplatform_postgres_readiness_or_data_source_postgres_ready")
         is True
     )
+    rollups["stockplatform_source_contract_readback_present"] = bool(
+        state.get("stockplatform_source_contract_readback_present") is True
+    )
+    rollups["stockplatform_source_contract_receipt_provided"] = bool(
+        state.get("stockplatform_source_contract_receipt_provided") is True
+    )
+    rollups["stockplatform_source_contract_ready"] = bool(
+        state.get("stockplatform_source_contract_ready") is True
+    )
+    rollups["stockplatform_source_contract_non_ok_source_count"] = _int(
+        state.get("stockplatform_source_contract_non_ok_source_count")
+    )
     rollups["stockplatform_freshness_status"] = str(
         state.get("stockplatform_freshness_status") or "unknown"
     )
@@ -3321,6 +3420,15 @@ def _refresh_rollups_after_stockplatform_overlay(
         state.get("stockplatform_postgres_readiness_or_data_source_receipt_provided")
         is True
     )
+    summary["stockplatform_source_contract_readback_status"] = str(
+        state.get("stockplatform_source_contract_readback_status") or "unknown"
+    )
+    summary["stockplatform_source_contract_receipt_provided"] = bool(
+        state.get("stockplatform_source_contract_receipt_provided") is True
+    )
+    summary["stockplatform_source_contract_non_ok_source_count"] = _int(
+        state.get("stockplatform_source_contract_non_ok_source_count")
+    )
     summary["stockplatform_freshness_status"] = str(
         state.get("stockplatform_freshness_status") or "unknown"
     )

apps/api/src/services/stockplatform_public_api_controlled_recovery_preflight.py

@@ -51,6 +51,9 @@ def _build_payload(
     data_readiness_control_readback = _dict(
         runtime.get("data_readiness_control_readback")
     )
+    source_contract_control_readback = _dict(
+        runtime.get("source_contract_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
@@ -128,10 +131,23 @@ def _build_payload(
             "data_readiness_control_receipt_provided": (
                 data_readiness_control_readback.get("receipt_provided") is True
             ),
+            "source_contract_control_readback_status": (
+                source_contract_control_readback.get("status")
+            ),
+            "source_contract_receipt_provided": (
+                source_contract_control_readback.get("receipt_provided") is True
+            ),
+            "source_contract_non_ok_source_count": _int(
+                source_contract_control_readback.get("non_ok_source_count")
+            ),
+            "source_contract_blocker_count": len(
+                _strings(source_contract_control_readback.get("active_blockers"))
+            ),
             "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,
+        "source_contract_control_readback": source_contract_control_readback,
         "controlled_apply_policy": {
             "risk_level": "high",
             "break_glass_required": False,
@@ -201,6 +217,21 @@ def _build_payload(
             "postgres_readiness_or_data_source_receipt_provided": (
                 data_readiness_control_readback.get("receipt_provided") is True
             ),
+            "source_contract_readback_present": (
+                source_contract_control_readback.get("readback_present") is True
+            ),
+            "source_contract_receipt_provided": (
+                source_contract_control_readback.get("receipt_provided") is True
+            ),
+            "source_contract_ready": (
+                source_contract_control_readback.get("source_contract_ready") is True
+            ),
+            "source_contract_non_ok_source_count": _int(
+                source_contract_control_readback.get("non_ok_source_count")
+            ),
+            "source_contract_blocker_count": len(
+                _strings(source_contract_control_readback.get("active_blockers"))
+            ),
         },
         "operation_boundaries": {
             "read_only_public_https_probe": True,

apps/api/src/services/stockplatform_public_api_runtime_readback.py

@@ -25,6 +25,8 @@ _DATA_READINESS_SCHEMA_VERSION = (
 _DATA_READINESS_RECEIPT_KEY = (
     "stockplatform_postgres_readiness_or_data_source_readback"
 )
+_SOURCE_CONTRACT_SCHEMA_VERSION = "stockplatform_source_contract_control_readback_v1"
+_SOURCE_CONTRACT_RECEIPT_KEY = "stockplatform_source_contract_readback"
 _DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
 _RECOVERY_RECEIPT_FILE = (
     "stockplatform-public-api-runtime-recovery-control-receipt.snapshot.json"
@@ -148,10 +150,20 @@ def _build_payload(
         postgres_not_ready=postgres_not_ready,
         data_dependency_blockers=active_blockers,
     )
+    source_contract_control_readback = _source_contract_control_readback(
+        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,
+        source_contract_control_readback=source_contract_control_readback,
     )
     recovery_control_path_blockers = _strings(
         recovery_control_path.get("active_blockers")
@@ -230,6 +242,22 @@ def _build_payload(
             "data_readiness_control_readback_present": (
                 data_readiness_control_readback.get("readback_present") is True
             ),
+            "source_contract_control_readback_status": (
+                source_contract_control_readback.get("status")
+            ),
+            "source_contract_control_readback_present": (
+                source_contract_control_readback.get("readback_present") is True
+            ),
+            "source_contract_receipt_provided": (
+                source_contract_control_readback.get("receipt_provided") is True
+            ),
+            "source_contract_non_ok_source_count": _int_or_none(
+                source_contract_control_readback.get("non_ok_source_count")
+            )
+            or 0,
+            "source_contract_blocker_count": len(
+                _strings(source_contract_control_readback.get("active_blockers"))
+            ),
         },
         "rollups": {
             "runtime_ready": ready,
@@ -266,8 +294,26 @@ def _build_payload(
             "postgres_readiness_or_data_source_receipt_provided": (
                 data_readiness_control_readback.get("receipt_provided") is True
             ),
+            "source_contract_readback_present": (
+                source_contract_control_readback.get("readback_present") is True
+            ),
+            "source_contract_receipt_provided": (
+                source_contract_control_readback.get("receipt_provided") is True
+            ),
+            "source_contract_ready": (
+                source_contract_control_readback.get("source_contract_ready")
+                is True
+            ),
+            "source_contract_non_ok_source_count": _int_or_none(
+                source_contract_control_readback.get("non_ok_source_count")
+            )
+            or 0,
+            "source_contract_blocker_count": len(
+                _strings(source_contract_control_readback.get("active_blockers"))
+            ),
         },
         "data_readiness_control_readback": data_readiness_control_readback,
+        "source_contract_control_readback": source_contract_control_readback,
         "recovery_control_path": recovery_control_path,
         "probes": probes,
         "operation_boundaries": {
@@ -293,6 +339,7 @@ def _recovery_control_path_readback(
     runtime_ready: bool,
     receipt: dict[str, Any],
     data_readiness_control_readback: dict[str, Any],
+    source_contract_control_readback: dict[str, Any],
 ) -> dict[str, Any]:
     if runtime_ready:
         return {
@@ -302,10 +349,12 @@ def _recovery_control_path_readback(
             "provided_receipts": _provided_receipts_with_live_readback(
                 _strings(receipt.get("provided_receipts")),
                 data_readiness_control_readback,
+                source_contract_control_readback,
             ),
             "missing_receipts": _missing_receipts_after_live_readback(
                 _strings(receipt.get("missing_receipts")),
                 data_readiness_control_readback,
+                source_contract_control_readback,
             ),
             "safe_recovery_channels": [
                 "keep_monitoring_public_api_runtime_readback",
@@ -315,10 +364,12 @@ def _recovery_control_path_readback(
         provided_receipts = _provided_receipts_with_live_readback(
             _strings(receipt.get("provided_receipts")),
             data_readiness_control_readback,
+            source_contract_control_readback,
         )
         missing_receipts = _missing_receipts_after_live_readback(
             _strings(receipt.get("missing_receipts")),
             data_readiness_control_readback,
+            source_contract_control_readback,
         )
         active_blockers = _recovery_control_active_blockers(
             receipt_blockers=_strings(receipt.get("active_blockers")),
@@ -339,6 +390,11 @@ def _recovery_control_path_readback(
                 if data_readiness_control_readback.get("receipt_provided") is True
                 else ""
             ),
+            "live_source_contract_receipt_key": (
+                _SOURCE_CONTRACT_RECEIPT_KEY
+                if source_contract_control_readback.get("receipt_provided") is True
+                else ""
+            ),
             "safe_recovery_channels": _strings(
                 receipt.get("safe_recovery_channels")
             ),
@@ -468,25 +524,151 @@ def _data_readiness_control_readback(
     }
 
 
+def _source_contract_control_readback(
+    *,
+    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
+    freshness_sources = _sources(freshness_json.get("sources"))
+    ingestion_sources = _sources(ingestion_json.get("sources"))
+    all_sources = [
+        *_source_entries(freshness_sources, "freshness"),
+        *_source_entries(ingestion_sources, "ingestion"),
+    ]
+    blocking_sources = _blocking_source_entries(
+        source_entries=all_sources,
+        freshness_blockers=_strings(freshness_json.get("blockers")),
+        ingestion_blockers=_strings(ingestion_json.get("blockers")),
+    )
+    data_ready = (
+        readback_present
+        and freshness_json.get("status") == "ok"
+        and ingestion_json.get("status") == "ok"
+    )
+    if data_ready:
+        status = "stockplatform_source_contract_ready"
+        safe_next_step = "run_post_apply_public_api_verifier_and_clear_p0_006_blocker"
+    elif readback_present and data_dependency_classification == (
+        "freshness_or_ingestion_not_ready"
+    ):
+        status = "blocked_stockplatform_source_contract_not_ready"
+        safe_next_step = (
+            "diff_stockplatform_source_contract_blocking_sources_then_rerun_"
+            "public_api_verifier"
+        )
+    elif readback_present and postgres_not_ready:
+        status = "stockplatform_source_contract_not_required_postgres_path"
+        safe_next_step = (
+            "run_stockplatform_production_migration_path_after_control_channel_"
+            "readback_no_manual_db_write"
+        )
+    elif readback_present:
+        status = "stockplatform_source_contract_review_required"
+        safe_next_step = "review_stockplatform_data_dependency_classification"
+    else:
+        status = "blocked_stockplatform_source_contract_readback_unavailable"
+        safe_next_step = (
+            "restore_public_api_freshness_ingestion_readback_then_rerun_"
+            "source_contract_readback"
+        )
+
+    active_blockers = [
+        blocker
+        for blocker in data_dependency_blockers
+        if blocker.startswith("stockplatform_freshness_")
+        or blocker.startswith("stockplatform_ingestion_")
+    ]
+    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")
+    if readback_present and not all_sources and not active_blockers and not data_ready:
+        missing_evidence.append("source_contract_sources_or_blockers")
+
+    return {
+        "schema_version": _SOURCE_CONTRACT_SCHEMA_VERSION,
+        "receipt_key": _SOURCE_CONTRACT_RECEIPT_KEY,
+        "status": status,
+        "source": "stockplatform_public_api_freshness_ingestion_source_contract",
+        "readback_present": readback_present,
+        "receipt_provided": readback_present,
+        "safe_next_step": safe_next_step,
+        "source_contract_ready": data_ready,
+        "data_dependency_classification": data_dependency_classification,
+        "postgres_not_ready": postgres_not_ready,
+        "freshness_status": str(freshness_json.get("status") or "unknown"),
+        "ingestion_status": str(ingestion_json.get("status") or "unknown"),
+        "latest_trading_date": str(
+            freshness_json.get("latest_trading_date")
+            or ingestion_json.get("latest_trading_date")
+            or ""
+        ),
+        "freshness_source_count": len(freshness_sources),
+        "ingestion_source_count": len(ingestion_sources),
+        "source_count": len(all_sources),
+        "sla_source_count": _sla_source_count(freshness_sources)
+        + _sla_source_count(ingestion_sources),
+        "non_ok_source_count": len(blocking_sources),
+        "source_contract_blockers": _unique_strings(
+            [
+                *(f"freshness:{blocker}" for blocker in _strings(freshness_json.get("blockers"))),
+                *(f"ingestion:{blocker}" for blocker in _strings(ingestion_json.get("blockers"))),
+            ]
+        ),
+        "blocking_sources": blocking_sources,
+        "active_blockers": active_blockers,
+        "missing_evidence": missing_evidence,
+        "post_apply_verifier_required": not data_ready,
+        "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],
+    source_contract_control_readback: dict[str, Any],
 ) -> list[str]:
+    receipts = list(provided_receipts)
     if data_readiness_control_readback.get("receipt_provided") is not True:
-        return provided_receipts
-    return _unique_strings([*provided_receipts, _DATA_READINESS_RECEIPT_KEY])
+        receipts = receipts
+    else:
+        receipts.append(_DATA_READINESS_RECEIPT_KEY)
+    if source_contract_control_readback.get("receipt_provided") is True:
+        receipts.append(_SOURCE_CONTRACT_RECEIPT_KEY)
+    return _unique_strings(receipts)
 
 
 def _missing_receipts_after_live_readback(
     missing_receipts: list[str],
     data_readiness_control_readback: dict[str, Any],
+    source_contract_control_readback: dict[str, Any],
 ) -> list[str]:
-    if data_readiness_control_readback.get("receipt_provided") is not True:
-        return missing_receipts
+    provided = set()
+    if data_readiness_control_readback.get("receipt_provided") is True:
+        provided.add(_DATA_READINESS_RECEIPT_KEY)
+    if source_contract_control_readback.get("receipt_provided") is True:
+        provided.add(_SOURCE_CONTRACT_RECEIPT_KEY)
     return [
         receipt
         for receipt in missing_receipts
-        if receipt != _DATA_READINESS_RECEIPT_KEY
+        if receipt not in provided
     ]
 
 
@@ -679,6 +861,77 @@ def _sources(value: Any) -> list[dict[str, Any]]:
     return [item for item in value if isinstance(item, dict)]
 
 
+def _source_entries(
+    sources: list[dict[str, Any]],
+    category: str,
+) -> list[dict[str, Any]]:
+    entries: list[dict[str, Any]] = []
+    for source in sources:
+        source_id = str(
+            source.get("source")
+            or source.get("name")
+            or source.get("id")
+            or source.get("table")
+            or "unknown"
+        )
+        status = str(source.get("status") or "unknown")
+        reason = str(
+            source.get("reason")
+            or source.get("message")
+            or source.get("blocker")
+            or ""
+        )
+        entries.append(
+            {
+                "source": source_id,
+                "category": category,
+                "status": status,
+                "reason": reason,
+                "has_sla": isinstance(source.get("sla"), dict),
+            }
+        )
+    return entries
+
+
+def _blocking_source_entries(
+    *,
+    source_entries: list[dict[str, Any]],
+    freshness_blockers: list[str],
+    ingestion_blockers: list[str],
+) -> list[dict[str, Any]]:
+    blocking = [
+        {
+            **entry,
+            "classification": f"{entry['category']}_source_non_ok",
+        }
+        for entry in source_entries
+        if entry.get("status") != "ok"
+    ]
+    if blocking:
+        return blocking
+    return [
+        {
+            "source": blocker,
+            "category": "freshness",
+            "status": "blocked",
+            "reason": blocker,
+            "has_sla": False,
+            "classification": "freshness_blocker_without_source_entry",
+        }
+        for blocker in freshness_blockers
+    ] + [
+        {
+            "source": blocker,
+            "category": "ingestion",
+            "status": "blocked",
+            "reason": blocker,
+            "has_sla": False,
+            "classification": "ingestion_blocker_without_source_entry",
+        }
+        for blocker in ingestion_blockers
+    ]
+
+
 def _sla_source_count(sources: list[dict[str, Any]]) -> int:
     return sum(1 for source in sources if isinstance(source.get("sla"), dict))