feat: separate live baseline continuity from candidate gate

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

platform-config.json

@@ -416,6 +416,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,
@@ -483,6 +491,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,
@@ -922,6 +936,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"}
@@ -299,6 +301,31 @@ def validate(config: dict) -> list[str]:
             errors.append(
                 f"strategy {sid}: live feature snapshot requires path and manifest_path"
             )
+        continuity = sdata.get("live_continuity")
+        if continuity is None:
+            continue
+        if not isinstance(continuity, dict):
+            errors.append(f"strategy {sid}: live_continuity must be an object")
+            continue
+        unsupported_continuity = sorted(set(continuity) - LIVE_CONTINUITY_POLICY_FIELDS)
+        if unsupported_continuity:
+            errors.append(
+                f"strategy {sid}: unsupported live_continuity fields {unsupported_continuity}"
+            )
+        eligible = continuity.get("eligible")
+        if not isinstance(eligible, bool):
+            errors.append(f"strategy {sid}: 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 {sid}: live_continuity.allowed_platforms must contain configured platforms"
+            )
+        elif len(set(allowed_platforms)) != len(allowed_platforms):
+            errors.append(f"strategy {sid}: live_continuity.allowed_platforms must not contain duplicates")
+        elif eligible is True and not allowed_platforms:
+            errors.append(f"strategy {sid}: live_continuity.eligible requires allowed_platforms")
     return errors
 
 
@@ -399,6 +426,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"]
@@ -438,6 +466,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")),
@@ -665,11 +697,16 @@ def _strategy_profile_gate_fields(sdata: dict) -> dict[str, object]:
     can_switch_live = sdata.get("can_switch_live", runtime_enabled and lifecycle_stage == "runtime_enabled")
     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/runtime_settings.py

@@ -4,13 +4,15 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
 import re
 import shlex
 import subprocess
 import sys
 from dataclasses import dataclass
+from datetime import date
 from pathlib import Path
 from typing import Any
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
@@ -53,6 +55,48 @@
 }
 SCHEDULER_FIELDS = frozenset({"timezone", "main_time", "probe_time", "precheck_time"})
 MARKET_FIELDS = ("market", "market_calendar", "market_timezone")
+STRATEGY_RELEASE_REQUIRED_FIELDS = (
+    "release_id",
+    "manifest_sha256",
+    "strategy_revision",
+    "config_sha256",
+    "risk_policy_sha256",
+    "evidence_sha256",
+    "plugin_bundle_sha256",
+    "effective_session",
+)
+STRATEGY_RELEASE_DIGEST_FIELDS = frozenset(
+    {
+        "manifest_sha256",
+        "config_sha256",
+        "risk_policy_sha256",
+        "evidence_sha256",
+        "plugin_bundle_sha256",
+    }
+)
+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(
@@ -165,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 every frozen target field except its current run 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
@@ -425,8 +478,102 @@ def validate_runtime_target(target: dict[str, Any], errors: list[str]) -> None:
             ZoneInfo(market_timezone)
         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:
+    """Validate an optional immutable release identity without enabling it.
 
+    Existing targets intentionally remain valid without ``strategy_release``
+    during the read-only migration.  Once present, however, a partial identity
+    is never accepted because it could be mistaken for a verified release.
+    """
+
+    release = runtime_target.get("strategy_release")
+    if release is None:
+        return
+    if not isinstance(release, dict):
+        errors.append("runtime_target.strategy_release must be an object when present")
+        return
+    unexpected = sorted(set(release) - set(STRATEGY_RELEASE_REQUIRED_FIELDS))
+    if unexpected:
+        errors.append(
+            "runtime_target.strategy_release contains unsupported fields: "
+            + ", ".join(unexpected)
+        )
+    for field in STRATEGY_RELEASE_REQUIRED_FIELDS:
+        value = release.get(field)
+        if not isinstance(value, str) or not value.strip():
+            errors.append(f"runtime_target.strategy_release.{field} is required")
+            continue
+        if field == "release_id" and not STRATEGY_RELEASE_ID_PATTERN.fullmatch(value.strip()):
+            errors.append("runtime_target.strategy_release.release_id has invalid characters")
+        if field in STRATEGY_RELEASE_DIGEST_FIELDS and not SHA256_PATTERN.fullmatch(value.strip()):
+            errors.append(f"runtime_target.strategy_release.{field} must be a SHA-256 digest")
+        if field == "effective_session":
+            try:
+                date.fromisoformat(value.strip())
+            except ValueError:
+                errors.append(
+                    "runtime_target.strategy_release.effective_session must be an ISO-8601 date"
+                )
 
+
+def validate_live_continuity(runtime_target: dict[str, Any], errors: list[str]) -> None:
+    """Validate a frozen incumbent baseline independently of P0--P6 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],
@@ -521,12 +668,26 @@ def validate_runtime_target_strategy_policy(runtime_target: dict[str, Any], erro
         and deployment.get("live_configured") is False
     ):
         errors.append(f"platform {platform_id} has no live runtime configuration")
+    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")