fix: fail closed and suppress non-execution heartbeats

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

Author: Pigbibi

decision_mapper.py

@@ -148,6 +148,14 @@ def map_strategy_decision(
         diagnostics["consecutive_losses"] = int(runtime_metadata["consecutive_losses"])
     risk_flags = tuple(str(flag) for flag in decision.risk_flags)
     no_execute = bool(_NO_EXECUTE_FLAGS & set(risk_flags))
+    if not no_execute and not decision.positions:
+        # An empty position set is not a safe implicit liquidation instruction.
+        # It can result from a degraded strategy plug-in or missing inputs, so
+        # keep the existing book unchanged until a strategy explicitly emits a
+        # releasable allocation.
+        no_execute = True
+        risk_flags = tuple(dict.fromkeys((*risk_flags, "no_execute")))
+        diagnostics.setdefault("execution_blocked_reason", "empty_position_decision")
     total_equity_value = runtime_metadata.get("portfolio_total_equity")
     cash_only_execution = bool(runtime_metadata.get("cash_only_execution", True))
     if not no_execute and total_equity_value is not None:

scripts/execution_report_heartbeat.py

@@ -22,6 +22,7 @@
         match_payload_target,
         runtime_target_configuration_has_enabled_targets,
         runtime_target_configuration_present,
+        runtime_target_permits_standard_execution,
         target_key,
         target_label,
         target_latest_due_at,
@@ -34,6 +35,7 @@
         match_payload_target,
         runtime_target_configuration_has_enabled_targets,
         runtime_target_configuration_present,
+        runtime_target_permits_standard_execution,
         target_key,
         target_label,
         target_latest_due_at,
@@ -124,7 +126,10 @@ def _target_enabled(target: dict[str, Any], runtime_target: dict[str, Any]) -> b
         value = target.get("RUNTIME_TARGET_ENABLED")
     if value is None:
         value = runtime_target.get("runtime_target_enabled")
-    return _enabled_value(value, default=True)
+    return (
+        _enabled_value(value, default=True)
+        and runtime_target_permits_standard_execution(runtime_target)
+    )
 
 
 def _target_service_values(target: dict[str, Any], runtime_target: dict[str, Any]) -> list[str]:

scripts/runtime_heartbeat_policy.py

@@ -49,6 +49,21 @@ def _enabled(value: Any, *, default: bool = True) -> bool:
     return str(value).strip().lower() not in {"0", "false", "no", "n", "off"}
 
 
+def runtime_target_permits_standard_execution(runtime_target: Mapping[str, Any]) -> bool:
+    """Return whether normal execution receipts are expected for a target.
+
+    A continuity record deliberately distinguishes an authorised live baseline
+    from reconciliation, pause, and specialised risk-reduction states. The
+    generic heartbeat must not demand a normal execution report when the
+    runtime is explicitly prohibited from normal execution; those states are
+    verified by their dedicated reconciliation controls instead.
+    """
+
+    continuity = _mapping(runtime_target.get("live_continuity"))
+    state = str(continuity.get("state") or "").strip().upper()
+    return not state or state in {"ACTIVE_LKG", "ROLLBACK_LKG"}
+
+
 def _mapping(value: Any) -> Mapping[str, Any]:
     return value if isinstance(value, Mapping) else {}
 
@@ -317,7 +332,7 @@ def runtime_target_configuration_has_enabled_targets(
         )
         if enabled_value is None:
             enabled_value = runtime_target.get("runtime_target_enabled")
-        if _enabled(enabled_value):
+        if _enabled(enabled_value) and runtime_target_permits_standard_execution(runtime_target):
             return True
     return False
 
@@ -342,7 +357,10 @@ def load_runtime_targets(
         )
         if enabled_value is None:
             enabled_value = runtime_target.get("runtime_target_enabled")
-        enabled = _enabled(enabled_value)
+        enabled = (
+            _enabled(enabled_value)
+            and runtime_target_permits_standard_execution(runtime_target)
+        )
         if not enabled and not include_disabled:
             continue
         target_scope = _first_value(

tests/test_decision_mapper.py

@@ -66,6 +66,25 @@ def test_map_strategy_decision_returns_noop_when_flagged_no_execute():
     assert "allocation" not in metadata
 
 
+def test_map_strategy_decision_fails_closed_for_unflagged_empty_positions():
+    target_weights, _signal_desc, _is_emergency, _status_desc, metadata = map_strategy_decision(
+        StrategyDecision(
+            diagnostics={
+                "signal_description": "strategy input unavailable",
+                "status_description": "waiting",
+            }
+        ),
+        strategy_profile="soxl_soxx_trend_income",
+        runtime_metadata={"managed_symbols": ("SOXL", "SOXX", "BOXX")},
+    )
+
+    assert target_weights is None
+    assert metadata["actionable"] is False
+    assert metadata["execution_blocked_reason"] == "empty_position_decision"
+    assert metadata["risk_flags"] == ("no_execute",)
+    assert "allocation" not in metadata
+
+
 def test_map_strategy_decision_translates_value_targets_for_semiconductor_strategy():
     decision = StrategyDecision(
         positions=(

tests/test_execution_report_heartbeat.py

@@ -98,6 +98,40 @@ def test_target_derived_required_services_skip_disabled_targets(monkeypatch):
     ]
 
 
+def test_target_derived_required_services_skip_reconcile_only_targets(monkeypatch):
+    monkeypatch.delenv("RUNTIME_HEARTBEAT_REQUIRED_SERVICES", raising=False)
+    monkeypatch.delenv("CLOUD_RUN_SERVICE", raising=False)
+    monkeypatch.delenv("CLOUD_RUN_SERVICES", raising=False)
+    monkeypatch.delenv("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE", raising=False)
+    monkeypatch.setenv(
+        "CLOUD_RUN_SERVICE_TARGETS_JSON",
+        json.dumps(
+            {
+                "targets": [
+                    {
+                        "service": "reconcile-only-service",
+                        "runtime_target": {
+                            "service_name": "reconcile-only-service",
+                            "strategy_profile": "strategy-a",
+                            "live_continuity": {"state": "RECONCILE_ONLY"},
+                        },
+                    },
+                    {
+                        "service": "active-service",
+                        "runtime_target": {
+                            "service_name": "active-service",
+                            "strategy_profile": "strategy-b",
+                            "live_continuity": {"state": "ACTIVE_LKG"},
+                        },
+                    },
+                ]
+            }
+        ),
+    )
+
+    assert heartbeat._load_required_services() == ["active-service"]
+
+
 def test_explicit_required_services_skip_disabled_targets(monkeypatch):
     monkeypatch.setenv(
         "RUNTIME_HEARTBEAT_REQUIRED_SERVICES",

tests/test_runtime_heartbeat_policy.py

@@ -276,6 +276,28 @@ def test_target_defaults_and_scheduler_aliases_are_normalized() -> None:
     }
 
 
+def test_reconcile_only_target_is_not_an_execution_heartbeat_target() -> None:
+    environ = {
+        "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(
+            {
+                "targets": [
+                    {
+                        "service": "reconcile-only-service",
+                        "runtime_target": {
+                            "service_name": "reconcile-only-service",
+                            "strategy_profile": "strategy-a",
+                            "live_continuity": {"state": "RECONCILE_ONLY"},
+                        },
+                    }
+                ]
+            }
+        )
+    }
+
+    assert load_runtime_targets(environ) == []
+    assert runtime_target_configuration_present(environ) is True
+
+
 
 def test_publication_grace_uses_previous_matured_schedule_cutoff() -> None:
     targets = load_runtime_targets(