Harden scheduler switch guards

Author: Pigbibi

.github/workflows/sync-cloud-run-env.yml

@@ -880,7 +880,8 @@ jobs:
           for target in plan["targets"]:
               service_name = str(target["service_name"]).strip()
               env = target.get("env") or {}
-              timezone = str(env.get("IBKR_MARKET_TIMEZONE") or "").strip()
+              scheduler = target.get("scheduler") or {}
+              timezone = str(scheduler.get("timezone") or env.get("IBKR_MARKET_TIMEZONE") or "").strip()
               market = str(env.get("IBKR_MARKET") or "").strip().upper()
               if not timezone:
                   timezone = "Asia/Hong_Kong" if market == "HK" else "America/New_York"
@@ -889,9 +890,9 @@ jobs:
                       [
                           service_name,
                           timezone,
-                          configured_time("CLOUD_SCHEDULER_MAIN_TIME", "45 15"),
-                          configured_time("CLOUD_SCHEDULER_PROBE_TIME", "35 9,15"),
-                          configured_time("CLOUD_SCHEDULER_PRECHECK_TIME", "45 9"),
+                          str(scheduler.get("main_time") or configured_time("CLOUD_SCHEDULER_MAIN_TIME", "45 15")),
+                          str(scheduler.get("probe_time") or configured_time("CLOUD_SCHEDULER_PROBE_TIME", "35 9,15")),
+                          str(scheduler.get("precheck_time") or configured_time("CLOUD_SCHEDULER_PRECHECK_TIME", "45 9")),
                       ]
                   )
               )

scripts/build_cloud_run_env_sync_plan.py

@@ -100,6 +100,16 @@ def _should_add_local_src(candidate: Path) -> bool:
     "IBKR_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD",
     "EXECUTION_REPORT_GCS_URI",
 )
+SCHEDULER_TIME_DEFAULTS = {
+    "main_time": "45 15",
+    "probe_time": "35 9,15",
+    "precheck_time": "45 9",
+}
+SCHEDULER_TIME_ENV = {
+    "main_time": "CLOUD_SCHEDULER_MAIN_TIME",
+    "probe_time": "CLOUD_SCHEDULER_PROBE_TIME",
+    "precheck_time": "CLOUD_SCHEDULER_PRECHECK_TIME",
+}
 
 
 def build_sync_plan(env: Mapping[str, str] = os.environ) -> dict[str, object]:
@@ -270,10 +280,46 @@ def _build_target_plan(
         "service_name": service_name,
         "strategy_profile": canonical_profile,
         "env": env_values,
+        "scheduler": _build_scheduler_plan(
+            target=target,
+            defaults=defaults,
+            env=env,
+            env_values=env_values,
+            per_service_mode=per_service_mode,
+        ),
         "remove_env_vars": sorted(set(remove_env_vars) - set(env_values)),
     }
 
 
+def _build_scheduler_plan(
+    *,
+    target: Mapping[str, object],
+    defaults: Mapping[str, object],
+    env: Mapping[str, str],
+    env_values: Mapping[str, str],
+    per_service_mode: bool,
+) -> dict[str, str]:
+    market = str(env_values.get("IBKR_MARKET") or "").strip().upper()
+    timezone = str(env_values.get("IBKR_MARKET_TIMEZONE") or "").strip()
+    if not timezone:
+        timezone = "Asia/Hong_Kong" if market == "HK" else "America/New_York"
+
+    scheduler = {"timezone": timezone}
+    for key, env_name in SCHEDULER_TIME_ENV.items():
+        scheduler[key] = (
+            _target_env_value(
+                target,
+                defaults,
+                env,
+                env_name,
+                per_service_mode=per_service_mode,
+                allow_shared_fallback=True,
+            )
+            or SCHEDULER_TIME_DEFAULTS[key]
+        )
+    return scheduler
+
+
 def _validate_profile_inputs(
     *,
     service_name: str,

scripts/cloud_run_runtime_guard.py

@@ -85,6 +85,19 @@ def _load_services() -> list[str]:
     return unique
 
 
+def _scheduler_job_pattern_for_services(services: list[str]) -> str:
+    candidates: list[str] = []
+    for service in services:
+        service_name = str(service or "").strip()
+        if not service_name:
+            continue
+        candidates.append(service_name)
+        if service_name.endswith("-service"):
+            candidates.append(service_name.removesuffix("-service"))
+    unique = list(dict.fromkeys(candidates))
+    return "|".join(re.escape(candidate) for candidate in unique)
+
+
 def _run_gcloud_logging(project: str, log_filter: str, limit: int) -> list[dict[str, Any]]:
     command = [
         "gcloud",
@@ -214,7 +227,6 @@ def main() -> int:
     require_success = _env_bool("RUNTIME_GUARD_REQUIRE_SUCCESS", False)
     fail_workflow = _env_bool("RUNTIME_GUARD_FAIL_WORKFLOW_ON_ALERT", True)
     check_scheduler = _env_bool("RUNTIME_GUARD_CHECK_SCHEDULER", True)
-    scheduler_pattern = os.environ.get("RUNTIME_GUARD_SCHEDULER_JOB_PATTERN") or ""
 
     since = (
         dt.datetime.now(dt.timezone.utc) - dt.timedelta(minutes=lookback_minutes)
@@ -230,6 +242,10 @@ def main() -> int:
     except RuntimeError as exc:
         services = []
         issues.append(f"service configuration error: {exc}")
+    scheduler_pattern = (
+        os.environ.get("RUNTIME_GUARD_SCHEDULER_JOB_PATTERN")
+        or _scheduler_job_pattern_for_services(services)
+    )
 
     for service in services:
         log_filter = (

scripts/execution_report_heartbeat.py

@@ -259,12 +259,24 @@ def _describe_scheduler_jobs_for_services(
 ) -> list[dict[str, Any]]:
     jobs = []
     for service in services:
-        job = _describe_scheduler_job(f"{service}-scheduler", project=project)
-        if job:
-            jobs.append(job)
+        for job_name in _scheduler_job_name_candidates(service):
+            job = _describe_scheduler_job(job_name, project=project)
+            if job:
+                jobs.append(job)
+                break
     return jobs
 
 
+def _scheduler_job_name_candidates(service: str) -> list[str]:
+    service_name = str(service or "").strip()
+    if not service_name:
+        return []
+    candidates = [f"{service_name}-scheduler"]
+    if service_name.endswith("-service"):
+        candidates.append(f"{service_name.removesuffix('-service')}-scheduler")
+    return _unique_values(candidates)
+
+
 def _scheduler_job_targets_strategy_run(job: dict[str, Any], service: str) -> bool:
     if str(job.get("state") or "").strip().upper() not in {"", "ENABLED"}:
         return False

tests/test_cloud_run_runtime_guard.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+import re
+
+from scripts import cloud_run_runtime_guard as guard
+
+
+def test_scheduler_job_pattern_includes_service_alias():
+    pattern = guard._scheduler_job_pattern_for_services(
+        ["interactive-brokers-live-u1599-tqqq-service"]
+    )
+
+    assert re.search(pattern, "interactive-brokers-live-u1599-tqqq-service-scheduler")
+    assert re.search(pattern, "interactive-brokers-live-u1599-tqqq-scheduler")
+    assert not re.search(pattern, "interactive-brokers-live-u1660-soxl-scheduler")

tests/test_execution_report_heartbeat.py

@@ -150,6 +150,46 @@ def test_scheduler_aware_required_services_fall_back_to_named_scheduler_describe
     assert scheduler_checked is True
 
 
+def test_scheduler_aware_named_fallback_uses_service_alias(monkeypatch):
+    monkeypatch.delenv("RUNTIME_HEARTBEAT_REQUIRED_SERVICES", raising=False)
+    monkeypatch.setenv("CLOUD_RUN_SERVICE", "interactive-brokers-live-u1599-tqqq-service")
+    monkeypatch.setattr(
+        heartbeat,
+        "_list_scheduler_jobs",
+        lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("cloudscheduler.jobs.list denied")),
+    )
+    requested_job_names = []
+
+    def fake_describe_scheduler_job(job_name, **_kwargs):
+        requested_job_names.append(job_name)
+        if job_name != "interactive-brokers-live-u1599-tqqq-scheduler":
+            return None
+        return {
+            "state": "ENABLED",
+            "schedule": "45 15 26 * *",
+            "timeZone": "America/New_York",
+            "httpTarget": {
+                "uri": "https://interactive-brokers-live-u1599-tqqq-service.example.run.app/"
+            },
+        }
+
+    monkeypatch.setattr(heartbeat, "_describe_scheduler_job", fake_describe_scheduler_job)
+
+    required, skip_reason, scheduler_checked = heartbeat._resolve_required_services(
+        project="project-1",
+        since=dt.datetime(2026, 6, 10, 0, 0, tzinfo=dt.timezone.utc),
+        now=dt.datetime(2026, 6, 10, 2, 0, tzinfo=dt.timezone.utc),
+    )
+
+    assert requested_job_names == [
+        "interactive-brokers-live-u1599-tqqq-service-scheduler",
+        "interactive-brokers-live-u1599-tqqq-scheduler",
+    ]
+    assert required == []
+    assert skip_reason and "no configured Cloud Scheduler main job was due" in skip_reason
+    assert scheduler_checked is True
+
+
 def test_main_skips_when_no_scheduler_main_job_is_due(monkeypatch, capsys):
     monkeypatch.delenv("RUNTIME_HEARTBEAT_REQUIRED_SERVICES", raising=False)
     monkeypatch.setenv("GCP_PROJECT_ID", "interactivebrokersquant")

tests/test_runtime_config_support.py

@@ -899,12 +899,14 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
             "IBKR_MARKET_DATA_SYMBOL_SUFFIX": ".HK",
             "IBKR_MARKET_EXCHANGE": "SEHK",
             "IBKR_MARKET_TIMEZONE": "Asia/Hong_Kong",
+            "cloud_scheduler_probe_time": "40 9,15",
             "EXECUTION_REPORT_GCS_URI": "gs://runtime/execution-reports",
         },
         "targets": [
             {
                 "service": "interactive-brokers-live-slot-a-service",
                 "account_group": "live-slot-a",
+                "cloud_scheduler_main_time": "10 16",
                 "runtime_target": json.loads(
                     runtime_target_json(
                         "tqqq_growth_income",
@@ -968,6 +970,12 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
     assert slot_a["env"]["IBKR_MARKET"] == "HK"
     assert slot_a["env"]["IBKR_MARKET_CURRENCY"] == "HKD"
     assert slot_a["env"]["IBKR_MARKET_EXCHANGE"] == "SEHK"
+    assert slot_a["scheduler"] == {
+        "timezone": "Asia/Hong_Kong",
+        "main_time": "10 16",
+        "probe_time": "40 9,15",
+        "precheck_time": "45 9",
+    }
     assert "IBKR_FEATURE_SNAPSHOT_PATH" not in slot_a["env"]
     assert "IBKR_FEATURE_SNAPSHOT_PATH" in slot_a["remove_env_vars"]
     assert "gs://stale-paper/snapshot.csv" not in json.dumps(slot_a)
@@ -977,6 +985,12 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
 
     assert u7654_mega["env"]["ACCOUNT_GROUP"] == "live-u7654-mega"
     assert u7654_mega["env"]["STRATEGY_PROFILE"] == "mega_cap_leader_rotation_top50_balanced"
+    assert u7654_mega["scheduler"] == {
+        "timezone": "Asia/Hong_Kong",
+        "main_time": "45 15",
+        "probe_time": "40 9,15",
+        "precheck_time": "45 9",
+    }
     assert u7654_mega["env"]["IBKR_FEATURE_SNAPSHOT_PATH"] == "gs://runtime/mega/snapshot.csv"
     assert (
         u7654_mega["env"]["IBKR_FEATURE_SNAPSHOT_MANIFEST_PATH"]

tests/test_sync_cloud_run_env_workflow.sh

@@ -88,7 +88,8 @@ grep -Fq -- '--remove-env-vars "$(IFS=,; echo "${remove_env_vars[*]}")' "$workfl
 grep -Fq -- '--update-env-vars "^|^$(join_by_delimiter "|" "${env_pairs[@]}")' "$workflow_file"
 grep -Fq 'Sync Cloud Scheduler schedule' "$workflow_file"
 grep -Fq 'scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"' "$workflow_file"
-grep -Fq 'timezone = str(env.get("IBKR_MARKET_TIMEZONE") or "").strip()' "$workflow_file"
+grep -Fq 'scheduler = target.get("scheduler") or {}' "$workflow_file"
+grep -Fq 'timezone = str(scheduler.get("timezone") or env.get("IBKR_MARKET_TIMEZONE") or "").strip()' "$workflow_file"
 grep -Fq 'timezone = "Asia/Hong_Kong" if market == "HK" else "America/New_York"' "$workflow_file"
 grep -Fq 'configured_time("CLOUD_SCHEDULER_MAIN_TIME", "45 15")' "$workflow_file"
 grep -Fq 'configured_time("CLOUD_SCHEDULER_PROBE_TIME", "35 9,15")' "$workflow_file"