feat: keep approved live champions separate from P0–P6 candidates

platform-config.json

@@ -426,6 +426,14 @@
         "dry_run"
       ],
       "blocked_live_reason": "missing_current_promotion_evidence_and_preauthorized_autonomy_policy",
+      "live_continuity": {
+        "eligible": true,
+        "allowed_platforms": [
+          "longbridge",
+          "ibkr",
+          "schwab"
+        ]
+      },
       "features": {
         "income_layer": true,
         "option_overlay": true,
@@ -493,6 +501,12 @@
         "dry_run"
       ],
       "blocked_live_reason": "missing_current_promotion_evidence_and_preauthorized_autonomy_policy",
+      "live_continuity": {
+        "eligible": true,
+        "allowed_platforms": [
+          "firstrade"
+        ]
+      },
       "features": {
         "income_layer": false,
         "option_overlay": false,
@@ -937,6 +951,12 @@
         "dry_run"
       ],
       "blocked_live_reason": "missing_current_promotion_evidence_and_preauthorized_autonomy_policy",
+      "live_continuity": {
+        "eligible": true,
+        "allowed_platforms": [
+          "binance"
+        ]
+      },
       "features": {
         "income_layer": false,
         "option_overlay": false,

python/scripts/build_config.py

@@ -37,7 +37,9 @@
     "can_switch_live",
     "allowed_execution_modes",
     "blocked_live_reason",
+    "live_continuity",
 }
+LIVE_CONTINUITY_POLICY_FIELDS = {"eligible", "allowed_platforms"}
 SCHEDULER_FIELDS = {"timezone", "main_time", "probe_time", "precheck_time"}
 MARKET_FIELDS = {"market", "market_calendar", "market_timezone"}
 FEATURE_SNAPSHOT_FIELDS = {"required", "path", "manifest_path", "max_age_days"}
@@ -94,6 +96,10 @@ def validate(config: dict) -> list[str]:
     errors: list[str] = []
     validate_runtime_authority_status(config, errors)
     validate_notification_references(config, errors)
+    platforms = config.get("platforms", {})
+    if not isinstance(platforms, dict):
+        errors.append("platforms must be an object")
+        platforms = {}
     scheduling = config.get("scheduling")
     scheduler_profiles = scheduling.get("profiles") if isinstance(scheduling, dict) else None
     if not isinstance(scheduler_profiles, dict) or not scheduler_profiles:
@@ -126,7 +132,7 @@ def validate(config: dict) -> list[str]:
                     errors.append(
                         f"scheduler profile {profile}: {field} must be Mon-Fri cron with day-of-month '*'"
                     )
-    for pid, pdata in config.get("platforms", {}).items():
+    for pid, pdata in platforms.items():
         if "capabilities" not in pdata:
             errors.append(f"platform {pid}: missing capabilities")
         if "default_account" not in pdata:
@@ -225,6 +231,7 @@ def validate(config: dict) -> list[str]:
                     f"must match market_timezone {market_timezone!r}"
                 )
     for sid, sdata in config.get("strategies", {}).items():
+        validate_live_continuity_policy(sid, sdata, platforms, errors)
         if "domain" not in sdata:
             errors.append(f"strategy {sid}: missing domain")
             continue
@@ -370,6 +377,41 @@ def validate(config: dict) -> list[str]:
     return errors
 
 
+def validate_live_continuity_policy(
+    strategy_id: str,
+    strategy: object,
+    platforms: dict,
+    errors: list[str],
+) -> None:
+    """Validate optional incumbent-continuity metadata independently of P0--P6."""
+
+    if not isinstance(strategy, dict):
+        return
+    continuity = strategy.get("live_continuity")
+    if continuity is None:
+        return
+    if not isinstance(continuity, dict):
+        errors.append(f"strategy {strategy_id}: live_continuity must be an object")
+        return
+    unsupported = sorted(set(continuity) - LIVE_CONTINUITY_POLICY_FIELDS)
+    if unsupported:
+        errors.append(f"strategy {strategy_id}: unsupported live_continuity fields {unsupported}")
+    eligible = continuity.get("eligible")
+    if not isinstance(eligible, bool):
+        errors.append(f"strategy {strategy_id}: live_continuity.eligible must be boolean")
+    allowed_platforms = continuity.get("allowed_platforms")
+    if not isinstance(allowed_platforms, list) or not all(
+        isinstance(platform, str) and platform in platforms for platform in allowed_platforms
+    ):
+        errors.append(
+            f"strategy {strategy_id}: live_continuity.allowed_platforms must contain configured platforms"
+        )
+    elif len(set(allowed_platforms)) != len(allowed_platforms):
+        errors.append(f"strategy {strategy_id}: live_continuity.allowed_platforms must not contain duplicates")
+    elif eligible is True and not allowed_platforms:
+        errors.append(f"strategy {strategy_id}: live_continuity.eligible requires allowed_platforms")
+
+
 def build_strategy_platform_dry_run_coverage(config: dict | None = None) -> dict[str, object]:
     """Report declared and default-buildable no-order platform routes.
 
@@ -698,6 +740,7 @@ def _automation_policy_for_strategy(profile: str, strategy: dict) -> dict[str, o
     runtime_enabled = strategy.get("runtime_enabled") is True
     blocked_reason = str(strategy.get("blocked_live_reason") or "").strip()
     features = strategy.get("features") if isinstance(strategy.get("features"), dict) else {}
+    continuity = strategy.get("live_continuity") if isinstance(strategy.get("live_continuity"), dict) else {}
     if runtime_enabled and can_switch_live and lifecycle_stage == "runtime_enabled":
         lane = "live_equivalent_optimization"
         triggers = ["health_degradation", "parameter_drift", "scheduled_retest", "market_regime_shift"]
@@ -737,6 +780,10 @@ def _automation_policy_for_strategy(profile: str, strategy: dict) -> dict[str, o
         "operating_policy_status": operating_policy_status,
         "can_switch_live": can_switch_live,
         "blocked_live_reason": blocked_reason,
+        "live_continuity": {
+            "eligible": continuity.get("eligible") is True,
+            "allowed_platforms": list(continuity.get("allowed_platforms") or []),
+        },
         "triggers": triggers,
         "evidence_required": evidence_required,
         "position_control_sensitive": bool(features.get("combo") or features.get("option_overlay")),
@@ -1005,11 +1052,16 @@ def _strategy_profile_gate_fields(sdata: dict) -> dict[str, object]:
     )
     if blocked_live_reason is None and not can_switch_live:
         blocked_live_reason = lifecycle_stage or "not_runtime_enabled"
+    continuity = sdata.get("live_continuity") if isinstance(sdata.get("live_continuity"), dict) else {}
     return {
         "lifecycle_stage": lifecycle_stage,
         "can_switch_live": can_switch_live,
         "allowed_execution_modes": _normalize_allowed_execution_modes(sdata.get("allowed_execution_modes")),
         "blocked_live_reason": "" if blocked_live_reason is None else str(blocked_live_reason).strip(),
+        "live_continuity": {
+            "eligible": continuity.get("eligible") is True,
+            "allowed_platforms": list(continuity.get("allowed_platforms") or []),
+        },
     }
 
 

python/scripts/build_platform_config.py

@@ -293,11 +293,16 @@ def _strategy_profile_gate_fields(sdata: dict) -> dict[str, object]:
     )
     if blocked_live_reason is None and not can_switch_live:
         blocked_live_reason = lifecycle_stage or "not_runtime_enabled"
+    continuity = sdata.get("live_continuity") if isinstance(sdata.get("live_continuity"), dict) else {}
     return {
         "lifecycle_stage": lifecycle_stage,
         "can_switch_live": can_switch_live,
         "allowed_execution_modes": _normalize_allowed_execution_modes(sdata.get("allowed_execution_modes")),
         "blocked_live_reason": "" if blocked_live_reason is None else str(blocked_live_reason).strip(),
+        "live_continuity": {
+            "eligible": continuity.get("eligible") is True,
+            "allowed_platforms": list(continuity.get("allowed_platforms") or []),
+        },
     }
 
 

python/scripts/inject_platform_config.py

@@ -158,6 +158,7 @@ def _strategy_profile_entry(sid: str, sdata: dict) -> dict:
     blocked_live_reason = sdata.get("blocked_live_reason")
     if blocked_live_reason is None and not can_switch_live:
         blocked_live_reason = lifecycle_stage or "not_runtime_enabled"
+    continuity = sdata.get("live_continuity") if isinstance(sdata.get("live_continuity"), dict) else {}
     entry = {
         "profile": sid,
         "label": sdata.get("label", sid),
@@ -169,6 +170,10 @@ def _strategy_profile_entry(sid: str, sdata: dict) -> dict:
         "can_switch_live": can_switch_live,
         "allowed_execution_modes": _normalize_allowed_execution_modes(sdata.get("allowed_execution_modes")),
         "blocked_live_reason": "" if blocked_live_reason is None else str(blocked_live_reason).strip(),
+        "live_continuity": {
+            "eligible": continuity.get("eligible") is True,
+            "allowed_platforms": list(continuity.get("allowed_platforms") or []),
+        },
         "income_layer_enabled": feat.get("income_layer", False),
         "option_overlay_enabled": feat.get("option_overlay", False),
         "combo_enabled": feat.get("combo", False),

python/scripts/runtime_settings.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
 import re
@@ -75,6 +76,27 @@
 )
 STRATEGY_RELEASE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{2,127}$")
 SHA256_PATTERN = re.compile(r"^(?:sha256:)?[0-9a-fA-F]{64}$")
+LIVE_CONTINUITY_STATES = frozenset(
+    {
+        "ACTIVE_LKG",
+        "ACTIVE_REDUCED",
+        "RECONCILE_ONLY",
+        "RISK_REDUCTION_ONLY",
+        "PAUSED",
+        "ROLLBACK_LKG",
+    }
+)
+LIVE_CONTINUITY_BASELINE_KINDS = frozenset({"legacy_authorized", "release_attested"})
+LIVE_CONTINUITY_FIELDS = frozenset(
+    {
+        "state",
+        "baseline_kind",
+        "baseline_id",
+        "baseline_target_sha256",
+        "captured_at",
+    }
+)
+LIVE_CONTINUITY_BASELINE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{2,127}$")
 GENERATED_VARIABLES = {"RUNTIME_TARGET_JSON", "STRATEGY_PROFILE"}
 SECRET_MARKERS = ("PASSWORD", "PRIVATE_KEY", "TOKEN", "API_KEY", "ACCESS_KEY", "CLIENT_SECRET", "SECRET")
 LEGACY_INCOME_LAYER_VARIABLES = frozenset(
@@ -187,6 +209,15 @@ def compact_json(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
 
 
+def runtime_target_fingerprint(runtime_target: dict[str, Any]) -> str:
+    """Fingerprint a frozen target excluding only its current continuity state."""
+
+    payload = dict(runtime_target)
+    payload.pop("live_continuity", None)
+    encoded = json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
 def env_string(value: Any) -> str:
     if isinstance(value, str):
         return value
@@ -453,6 +484,7 @@ def validate_runtime_target(target: dict[str, Any], errors: list[str]) -> None:
         except (ZoneInfoNotFoundError, ValueError):
             errors.append(f"runtime_target.market_timezone is invalid: {market_timezone!r}")
     validate_strategy_release(runtime_target, errors)
+    validate_live_continuity(runtime_target, errors)
 
 
 def validate_strategy_release(runtime_target: dict[str, Any], errors: list[str]) -> None:
@@ -493,6 +525,62 @@ def validate_strategy_release(runtime_target: dict[str, Any], errors: list[str])
                 )
 
 
+def validate_live_continuity(runtime_target: dict[str, Any], errors: list[str]) -> None:
+    """Validate a frozen incumbent baseline independently of candidate policy."""
+
+    continuity = runtime_target.get("live_continuity")
+    if continuity is None:
+        return
+    if not isinstance(continuity, dict):
+        errors.append("runtime_target.live_continuity must be an object when present")
+        return
+    unsupported = sorted(set(continuity) - LIVE_CONTINUITY_FIELDS)
+    if unsupported:
+        errors.append(
+            "runtime_target.live_continuity contains unsupported fields: " + ", ".join(unsupported)
+        )
+    for field in sorted(LIVE_CONTINUITY_FIELDS):
+        value = continuity.get(field)
+        if not isinstance(value, str) or not value.strip():
+            errors.append(f"runtime_target.live_continuity.{field} is required")
+
+    state = str(continuity.get("state") or "").strip().upper()
+    if state not in LIVE_CONTINUITY_STATES:
+        errors.append(
+            "runtime_target.live_continuity.state must be one of "
+            + ", ".join(sorted(LIVE_CONTINUITY_STATES))
+        )
+    baseline_kind = str(continuity.get("baseline_kind") or "").strip()
+    if baseline_kind not in LIVE_CONTINUITY_BASELINE_KINDS:
+        errors.append(
+            "runtime_target.live_continuity.baseline_kind must be one of "
+            + ", ".join(sorted(LIVE_CONTINUITY_BASELINE_KINDS))
+        )
+    baseline_id = str(continuity.get("baseline_id") or "").strip()
+    if baseline_id and not LIVE_CONTINUITY_BASELINE_ID_PATTERN.fullmatch(baseline_id):
+        errors.append("runtime_target.live_continuity.baseline_id has invalid characters")
+    digest = str(continuity.get("baseline_target_sha256") or "").strip().lower()
+    digest = digest.removeprefix("sha256:")
+    if not SHA256_PATTERN.fullmatch(digest):
+        errors.append("runtime_target.live_continuity.baseline_target_sha256 must be a SHA-256 digest")
+    elif digest != runtime_target_fingerprint(runtime_target):
+        errors.append(
+            "runtime_target.live_continuity.baseline_target_sha256 does not match the runtime target"
+        )
+    captured_at = str(continuity.get("captured_at") or "").strip()
+    if captured_at:
+        try:
+            date.fromisoformat(captured_at)
+        except ValueError:
+            errors.append("runtime_target.live_continuity.captured_at must be an ISO-8601 date")
+    if baseline_kind == "release_attested" and not isinstance(runtime_target.get("strategy_release"), dict):
+        errors.append("release_attested live_continuity requires runtime_target.strategy_release")
+
+
+def is_live_continuity_target(runtime_target: dict[str, Any]) -> bool:
+    return isinstance(runtime_target.get("live_continuity"), dict)
+
+
 def validate_live_ibkr_us_scheduler(
     runtime_target: dict[str, Any],
     scheduler: dict[str, Any],
@@ -596,12 +684,26 @@ def validate_runtime_target_strategy_policy(runtime_target: dict[str, Any], erro
         errors.append(
             f"platform {platform_id} does not support {execution_mode} control execution"
         )
+    continuity_target = is_live_continuity_target(runtime_target)
     allowed_modes = normalize_allowed_execution_modes(strategy.get("allowed_execution_modes"))
-    if allowed_modes and execution_mode not in allowed_modes:
+    if allowed_modes and execution_mode not in allowed_modes and not continuity_target:
         errors.append(f"runtime_target.strategy_profile {profile} does not allow {execution_mode} execution")
 
     if execution_mode != "live":
         return
+    if continuity_target:
+        continuity_policy = strategy.get("live_continuity")
+        if not isinstance(continuity_policy, dict) or continuity_policy.get("eligible") is not True:
+            errors.append(
+                f"runtime_target.strategy_profile {profile} is not eligible for live continuity"
+            )
+            return
+        allowed_platforms = continuity_policy.get("allowed_platforms")
+        if not isinstance(allowed_platforms, list) or platform_id not in allowed_platforms:
+            errors.append(
+                f"runtime_target.strategy_profile {profile} live continuity is not allowed on {platform_id}"
+            )
+        return
     lifecycle_stage = str(strategy.get("lifecycle_stage") or "").strip()
     if strategy.get("runtime_enabled") is not True:
         errors.append(f"runtime_target.strategy_profile {profile} is not runtime_enabled")

python/scripts/sync_strategy_switch_page_asset.py

@@ -81,6 +81,15 @@ def _enrich_profiles_from_config(profiles: list[dict]) -> list[dict]:
             "blocked_live_reason",
             "" if blocked_live_reason is None else str(blocked_live_reason).strip(),
         )
+        continuity = config_fields.get("live_continuity")
+        if isinstance(continuity, dict):
+            item.setdefault(
+                "live_continuity",
+                {
+                    "eligible": continuity.get("eligible") is True,
+                    "allowed_platforms": list(continuity.get("allowed_platforms") or []),
+                },
+            )
         item.setdefault("runtime_enabled", runtime_enabled)
         enriched.append(item)
     return enriched