A6-R2E-PAY8-D2-B Flagship trial runtime closure

core/generation_runner.py

@@ -4160,11 +4160,35 @@ class GenerationRunner:
         entitlement_class = str(snapshot.get("entitlement_class") or "").strip().upper()
         if tier not in {"STANDARD", "FLAGSHIP"}:
             raise RuntimeError("trusted_task_policy_product_tier_invalid")
-        if entitlement_class not in {"STANDARD_TRIAL", "STANDARD_PAID", "FLAGSHIP_PAID"}:
+
+        # Keep runner validation aligned with the executor's trusted-policy
+        # contract.  Entitlement authority comes from the SHA-bound resolved
+        # ProductPolicy, not from a duplicated runtime enum.  This preserves
+        # fail-closed behavior while allowing FLAGSHIP_TRIAL when (and only
+        # when) the trusted trial policy explicitly authorizes it.
+        trial = resolved.get("trial") if isinstance(resolved.get("trial"), dict) else None
+        billing = resolved.get("billing") if isinstance(resolved.get("billing"), dict) else None
+        allowed_entitlements: list[str] = []
+        if trial and str(trial.get("entitlement_class") or "").strip():
+            allowed_entitlements.append(str(trial.get("entitlement_class") or "").strip().upper())
+        if billing and str(billing.get("entitlement_class") or "").strip():
+            paid_class = str(billing.get("entitlement_class") or "").strip().upper()
+            if paid_class not in allowed_entitlements:
+                allowed_entitlements.append(paid_class)
+
+        # Backward-compatible trusted snapshots created before the nested
+        # trial/billing objects were embedded still carry SHA-bound policy
+        # version links.  Derive their same-tier classes from those links
+        # rather than reopening a duplicated global entitlement whitelist.
+        if not allowed_entitlements:
+            if str(resolved.get("trial_policy_version") or "").strip():
+                allowed_entitlements.append(f"{tier}_TRIAL")
+            if str(resolved.get("billing_policy_version") or "").strip():
+                allowed_entitlements.append(f"{tier}_PAID")
+
+        if not entitlement_class or entitlement_class not in allowed_entitlements:
             raise RuntimeError("trusted_task_policy_entitlement_invalid")
-        if tier == "STANDARD" and entitlement_class == "FLAGSHIP_PAID":
-            raise RuntimeError("trusted_task_policy_entitlement_mismatch")
-        if tier == "FLAGSHIP" and entitlement_class != "FLAGSHIP_PAID":
+        if any(not value.startswith(f"{tier}_") for value in allowed_entitlements):
             raise RuntimeError("trusted_task_policy_entitlement_mismatch")
         if str(resolved.get("product_tier") or "").strip().upper() != tier:
             raise RuntimeError("trusted_task_policy_product_tier_mismatch")