Fall back to named scheduler lookup

scripts/execution_report_heartbeat.py

@@ -278,6 +278,47 @@ def _list_scheduler_jobs(*, project: str | None) -> list[dict[str, Any]]:
     return payload if isinstance(payload, list) else []
 
 
+def _describe_scheduler_job(job_name: str, *, project: str | None) -> dict[str, Any] | None:
+    command = [
+        "gcloud",
+        "scheduler",
+        "jobs",
+        "describe",
+        job_name,
+        "--location",
+        _scheduler_location(),
+        "--format=json",
+    ]
+    if project:
+        command.extend(["--project", project])
+    result = _run_gcloud(command)
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        if "not_found" in detail.lower() or "not found" in detail.lower():
+            return None
+        raise RuntimeError(detail or f"gcloud scheduler jobs describe failed for {job_name}")
+    if not result.stdout.strip():
+        return None
+    try:
+        payload = json.loads(result.stdout)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(f"gcloud scheduler jobs describe returned invalid JSON for {job_name}: {exc}") from exc
+    return payload if isinstance(payload, dict) else None
+
+
+def _describe_scheduler_jobs_for_services(
+    services: list[str],
+    *,
+    project: str | None,
+) -> list[dict[str, Any]]:
+    jobs = []
+    for service in services:
+        job = _describe_scheduler_job(f"{service}-scheduler", project=project)
+        if job:
+            jobs.append(job)
+    return jobs
+
+
 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
@@ -401,7 +442,14 @@ def _filter_scheduler_due_services(
     since: dt.datetime,
     now: dt.datetime,
 ) -> list[str]:
-    jobs = _list_scheduler_jobs(project=project)
+    try:
+        jobs = _list_scheduler_jobs(project=project)
+    except RuntimeError as exc:
+        print(
+            f"Unable to list Cloud Scheduler jobs: {exc}; trying named scheduler job lookup.",
+            file=sys.stderr,
+        )
+        jobs = _describe_scheduler_jobs_for_services(services, project=project)
     due_services = []
     for service in services:
         service_jobs = [

tests/test_execution_report_heartbeat.py

@@ -140,6 +140,38 @@ def test_scheduler_aware_required_services_include_monthly_service_when_due(monk
     assert required == ["svc-monthly"]
 
 
+def test_scheduler_aware_required_services_fall_back_to_named_scheduler_describe(monkeypatch):
+    _clear_runtime_env(monkeypatch)
+    monkeypatch.setenv("CLOUD_RUN_SERVICE", "svc-monthly")
+    monkeypatch.setattr(
+        heartbeat,
+        "_list_scheduler_jobs",
+        lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("cloudscheduler.jobs.list denied")),
+    )
+    monkeypatch.setattr(
+        heartbeat,
+        "_describe_scheduler_job",
+        lambda job_name, **_kwargs: {
+            "state": "ENABLED",
+            "schedule": "45 15 1-7 * *",
+            "timeZone": "America/New_York",
+            "httpTarget": {"uri": "https://svc-monthly.example.run.app/"},
+        }
+        if job_name == "svc-monthly-scheduler"
+        else None,
+    )
+
+    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 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):
     _clear_runtime_env(monkeypatch)
     monkeypatch.setenv("GCP_PROJECT_ID", "longbridgequant")