|
32 | 32 | "service_name", |
33 | 33 | "execution_mode", |
34 | 34 | ) |
| 35 | +WINDOW_MODES = { |
| 36 | + "precheck": {"notify_only", "dry_run"}, |
| 37 | + "execution": {"live", "paper", "dry_run"}, |
| 38 | +} |
35 | 39 | GENERATED_VARIABLES = {"RUNTIME_TARGET_JSON", "STRATEGY_PROFILE"} |
36 | 40 | SECRET_MARKERS = ("PASSWORD", "PRIVATE_KEY", "TOKEN", "API_KEY") |
37 | 41 |
|
@@ -168,6 +172,45 @@ def validate_runtime_target(target: dict[str, Any], errors: list[str]) -> None: |
168 | 172 | if execution_mode not in {"live", "paper", "dry_run"}: |
169 | 173 | errors.append("runtime_target.execution_mode must be live, paper, or dry_run") |
170 | 174 |
|
| 175 | + execution_windows = runtime_target.get("execution_windows") |
| 176 | + if execution_windows is not None: |
| 177 | + if not isinstance(execution_windows, dict): |
| 178 | + errors.append("runtime_target.execution_windows must be an object when present") |
| 179 | + else: |
| 180 | + for window_name, allowed_modes in WINDOW_MODES.items(): |
| 181 | + window = execution_windows.get(window_name) |
| 182 | + if window is None: |
| 183 | + continue |
| 184 | + if not isinstance(window, dict): |
| 185 | + errors.append(f"runtime_target.execution_windows.{window_name} must be an object") |
| 186 | + continue |
| 187 | + for field in window: |
| 188 | + if field not in {"enabled", "offset_minutes", "mode"}: |
| 189 | + errors.append( |
| 190 | + f"runtime_target.execution_windows.{window_name}.{field} is unsupported" |
| 191 | + ) |
| 192 | + if "enabled" in window and not isinstance(window["enabled"], bool): |
| 193 | + errors.append( |
| 194 | + f"runtime_target.execution_windows.{window_name}.enabled must be boolean" |
| 195 | + ) |
| 196 | + if "offset_minutes" in window: |
| 197 | + offset_minutes = window["offset_minutes"] |
| 198 | + if not isinstance(offset_minutes, int) or offset_minutes < 0: |
| 199 | + errors.append( |
| 200 | + f"runtime_target.execution_windows.{window_name}.offset_minutes must be a non-negative integer" |
| 201 | + ) |
| 202 | + mode = window.get("mode") |
| 203 | + if mode is not None and mode not in allowed_modes: |
| 204 | + errors.append( |
| 205 | + f"runtime_target.execution_windows.{window_name}.mode must be one of {sorted(allowed_modes)}" |
| 206 | + ) |
| 207 | + for window_name in execution_windows: |
| 208 | + if window_name not in WINDOW_MODES: |
| 209 | + errors.append( |
| 210 | + "runtime_target.execution_windows only supports precheck and execution" |
| 211 | + ) |
| 212 | + break |
| 213 | + |
171 | 214 |
|
172 | 215 | def validate_plugin_mounts(target: dict[str, Any], errors: list[str]) -> None: |
173 | 216 | runtime_target = target.get("runtime_target") if isinstance(target.get("runtime_target"), dict) else {} |
|
0 commit comments