diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml index f7760ba..5c6e5f9 100644 --- a/.github/workflows/execution-report-heartbeat.yml +++ b/.github/workflows/execution-report-heartbeat.yml @@ -1,6 +1,5 @@ name: Execution Report Heartbeat -# Schedule disabled; trade/error notifications own unattended alerting. on: workflow_dispatch: inputs: @@ -17,6 +16,8 @@ on: options: - "true" - "false" + schedule: + - cron: "20 22 * * *" env: GCP_PROJECT_ID: firstradequant @@ -41,8 +42,20 @@ jobs: RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT: ${{ inputs.fail_workflow_on_alert || vars.RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT || 'true' }} RUNTIME_HEARTBEAT_ACCEPT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STAGES }} RUNTIME_HEARTBEAT_REJECT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STAGES }} + RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }} + RUNTIME_HEARTBEAT_MARKET_CALENDAR: ${{ vars.FIRSTRADE_MARKET_CALENDAR }} + RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.FIRSTRADE_MARKET_TIMEZONE }} + RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES: ${{ vars.RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES || '30' }} + RUNTIME_HEARTBEAT_SCHEDULER_AWARE: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_AWARE || 'true' }} + RUNTIME_HEARTBEAT_SCHEDULER_LOCATION: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_LOCATION || vars.CLOUD_RUN_REGION || 'us-central1' }} RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }} RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON }} + CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} + CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }} + CLOUD_RUN_SERVICES: ${{ vars.CLOUD_RUN_SERVICES }} + CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }} + CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }} + EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI }} FIRSTRADE_GCS_STATE_BUCKET: ${{ vars.FIRSTRADE_GCS_STATE_BUCKET }} FIRSTRADE_STATE_PREFIX: ${{ vars.FIRSTRADE_STATE_PREFIX }} GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} @@ -53,6 +66,19 @@ jobs: uses: actions/checkout@v6 - name: Authenticate to Google Cloud + id: gcp_auth_primary + continue-on-error: true + uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + + - name: Wait before Google Cloud authentication retry + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} + run: sleep 10 + + - name: Retry Google Cloud authentication + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} uses: google-github-actions/auth@v3 with: workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} @@ -61,5 +87,11 @@ jobs: - name: Set up gcloud uses: google-github-actions/setup-gcloud@v3 + - name: Install market calendar + continue-on-error: true + run: >- + python -m pip install --disable-pip-version-check + --retries 3 --timeout 30 "pandas-market-calendars==5.4.0" + - name: Check recent execution report run: python scripts/execution_report_heartbeat.py diff --git a/.github/workflows/runtime-guard.yml b/.github/workflows/runtime-guard.yml index c8a7eae..01bf065 100644 --- a/.github/workflows/runtime-guard.yml +++ b/.github/workflows/runtime-guard.yml @@ -63,6 +63,19 @@ jobs: uses: actions/checkout@v6 - name: Authenticate to Google Cloud + id: gcp_auth_primary + continue-on-error: true + uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + + - name: Wait before Google Cloud authentication retry + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} + run: sleep 10 + + - name: Retry Google Cloud authentication + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} uses: google-github-actions/auth@v3 with: workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 9f8b94e..1b6d3a2 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -47,6 +47,7 @@ jobs: QSL_ENABLE_CLOUD_RUN_AUTOMATION: ${{ vars.QSL_ENABLE_CLOUD_RUN_AUTOMATION }} CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }} + CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }} CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }} CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }} CLOUD_SCHEDULER_PROBE_TIME: ${{ vars.CLOUD_SCHEDULER_PROBE_TIME }} diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 4d61f2c..aee3fc9 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -242,11 +242,19 @@ def publish_log(text: str) -> None: except TypeError: log_message(text) + delivery_sent = True + + def send_and_capture(text: str) -> bool | None: + nonlocal delivery_sent + outcome = sender(text) + delivery_sent = outcome is not False + return outcome + NotificationPublisher( log_message=publish_log, - send_message=sender, + send_message=send_and_capture, ).publish(RenderedNotification(detailed_text=message, compact_text=message)) - return True + return delivery_sent def _should_publish_cycle_notification(result: Mapping[str, Any]) -> bool: @@ -733,6 +741,8 @@ def log_message(message: str) -> None: settings=settings, notification_sender=notification_sender, ) + if not result["notification_sent"]: + result["notification_error"] = "delivery_not_acknowledged" except Exception as exc: result["notification_sent"] = False result["notification_error"] = f"{type(exc).__name__}: {exc}" diff --git a/notifications/telegram.py b/notifications/telegram.py index 8a7e6d9..97c66f9 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -553,18 +553,29 @@ def build_sender(token: str | None, chat_id: str | None, *, requests_module=None if requests_module is None: import requests as requests_module - def send_tg_message(message: str) -> None: + def send_tg_message(message: str) -> bool: if not token or not chat_id: - return + return False url = f"https://api.telegram.org/bot{token}/sendMessage" try: - requests_module.post( + response = requests_module.post( url, json={"chat_id": chat_id, "text": _break_telegram_market_symbol_auto_links(message)}, timeout=15, ) + status_code = int(getattr(response, "status_code", 200) or 200) + if status_code < 200 or status_code >= 300: + print(f"Telegram send failed: HTTP {status_code}", flush=True) + return False + load_payload = getattr(response, "json", None) + payload = load_payload() if callable(load_payload) else None + if isinstance(payload, Mapping) and payload.get("ok") is False: + print("Telegram send failed: negative API acknowledgement", flush=True) + return False except Exception as exc: print(f"Telegram send failed: {type(exc).__name__}", flush=True) + return False + return True return send_tg_message diff --git a/scripts/cloud_run_runtime_guard.py b/scripts/cloud_run_runtime_guard.py index 7a1768c..7bc893a 100644 --- a/scripts/cloud_run_runtime_guard.py +++ b/scripts/cloud_run_runtime_guard.py @@ -23,6 +23,7 @@ "URL_ERROR", "URL_UNREACHABLE", ) +SCHEDULER_CLOUD_RUN_DEDUP_SECONDS = 120 def _split_values(raw: str | None) -> list[str]: @@ -40,6 +41,8 @@ def _env_bool(name: str, default: bool = False) -> bool: def _load_services() -> list[str]: services = [] + enabled_target_services = [] + disabled_target_services = [] for name in ( "RUNTIME_GUARD_CLOUD_RUN_SERVICES", "CLOUD_RUN_SERVICES", @@ -51,37 +54,27 @@ def _load_services() -> list[str]: if raw_targets: try: payload = json.loads(raw_targets) + defaults = payload.get("defaults") if isinstance(payload, dict) else {} + defaults = defaults if isinstance(defaults, dict) else {} targets = payload.get("targets") if isinstance(payload, dict) else payload if isinstance(targets, list): for target in targets: if not isinstance(target, dict): continue - if not _target_enabled(target): - continue - runtime_target = target.get("runtime_target") or target.get( - "runtime_target_json" - ) - if isinstance(runtime_target, str): - try: - runtime_target = json.loads(runtime_target) - except json.JSONDecodeError: - runtime_target = {} - for key in ("service", "service_name", "cloud_run_service"): - value = target.get(key) or ( - runtime_target.get(key) - if isinstance(runtime_target, dict) - else None - ) - if value: - services.extend(_split_values(str(value))) - break + target_services = _target_service_names(target, defaults) + if _target_enabled(target, defaults): + enabled_target_services.extend(target_services) + else: + disabled_target_services.extend(target_services) except json.JSONDecodeError as exc: raise RuntimeError(f"CLOUD_RUN_SERVICE_TARGETS_JSON is invalid: {exc}") from exc + services.extend(enabled_target_services) + disabled = set(disabled_target_services) - set(enabled_target_services) seen = set() unique = [] for service in services: - if service not in seen: + if service not in seen and service not in disabled: seen.add(service) unique.append(service) return unique @@ -111,9 +104,29 @@ def _service_job_aliases(service: str) -> list[str]: def _scheduler_job_pattern_for_services(services: list[str]) -> str: candidates: list[str] = [] for service in services: - candidates.extend(_service_job_aliases(service)) + candidates.extend(_scheduler_job_names(service)) unique = list(dict.fromkeys(candidates)) - return "|".join(re.escape(candidate) for candidate in unique) + if not unique: + return "" + return r"^(?:" + "|".join(re.escape(candidate) for candidate in unique) + r")\Z" + + +def _scheduler_job_names(service: str) -> list[str]: + names = [] + for alias in _service_job_aliases(service): + names.extend( + ( + f"{alias}-scheduler", + f"{alias}-probe-scheduler", + f"{alias}-precheck-scheduler", + ) + ) + return list(dict.fromkeys(names)) + + +def _job_matches_service(job_name: str, service: str) -> bool: + normalized = str(job_name or "").strip().rsplit("/", 1)[-1] + return normalized in _scheduler_job_names(service) def _entry_job_name(entry: dict[str, Any]) -> str: @@ -130,11 +143,46 @@ def _scheduler_entry_since( matches = [ service_since for service, service_since in service_since_by_name.items() - if any(alias and alias in job_name for alias in _service_job_aliases(service)) + if _job_matches_service(job_name, service) ] return max(matches) if matches else fallback +def _is_duplicate_scheduler_failure( + entry: dict[str, Any], + cloud_run_failures_by_service: dict[str, list[dict[str, Any]]], +) -> bool: + scheduler_timestamp = _parse_timestamp(entry.get("timestamp")) + job_name = _entry_job_name(entry) + if scheduler_timestamp is None or not job_name: + return False + + tolerance = dt.timedelta(seconds=SCHEDULER_CLOUD_RUN_DEDUP_SECONDS) + for service, failures in cloud_run_failures_by_service.items(): + if not _job_matches_service(job_name, service): + continue + for failure in failures: + cloud_run_timestamp = _parse_timestamp(failure.get("timestamp")) + if ( + cloud_run_timestamp is not None + and abs(scheduler_timestamp - cloud_run_timestamp) <= tolerance + ): + return True + return False + + +def _services_without_success( + services: list[str], + success_count_by_service: dict[str, int], + queried_services: set[str], +) -> list[str]: + return [ + service + for service in services + if service in queried_services and success_count_by_service.get(service, 0) == 0 + ] + + def _run_gcloud(args: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(args, text=True, capture_output=True, check=False) @@ -197,22 +245,51 @@ def _format_timestamp(value: dt.datetime) -> str: return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") -def _target_payloads() -> list[dict[str, Any]]: +def _target_configuration() -> tuple[list[dict[str, Any]], dict[str, Any]]: raw_targets = (os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() if not raw_targets: - return [] + return [], {} try: payload = json.loads(raw_targets) except json.JSONDecodeError: - return [] + return [], {} + defaults = payload.get("defaults") if isinstance(payload, dict) else {} + defaults = defaults if isinstance(defaults, dict) else {} targets = payload.get("targets") if isinstance(payload, dict) else payload if not isinstance(targets, list): - return [] - return [target for target in targets if isinstance(target, dict)] + return [], defaults + return [target for target in targets if isinstance(target, dict)], defaults -def _runtime_target(target: dict[str, Any]) -> dict[str, Any]: - runtime_target = target.get("runtime_target") or target.get("runtime_target_json") +def _target_payloads() -> list[dict[str, Any]]: + targets, _defaults = _target_configuration() + return targets + + +def _target_field( + target: dict[str, Any], + defaults: dict[str, Any], + *names: str, +) -> Any: + target_env = target.get("env") if isinstance(target.get("env"), dict) else {} + defaults_env = defaults.get("env") if isinstance(defaults.get("env"), dict) else {} + for source in (target, target_env, defaults, defaults_env): + for name in names: + if name in source: + return source[name] + return None + + +def _runtime_target( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> dict[str, Any]: + runtime_target = _target_field( + target, + defaults or {}, + "runtime_target", + "runtime_target_json", + ) if isinstance(runtime_target, str): try: runtime_target = json.loads(runtime_target) @@ -232,32 +309,57 @@ def _coerce_bool(value: Any, default: bool) -> bool: return text in {"1", "true", "yes", "y", "on"} -def _target_enabled(target: dict[str, Any]) -> bool: - runtime_target = _runtime_target(target) +def _target_enabled( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> bool: + defaults = defaults or {} + runtime_target = _runtime_target(target, defaults) + value = _target_field( + target, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) + if value is not None: + return _coerce_bool(value, True) for key in ("runtime_target_enabled", "RUNTIME_TARGET_ENABLED"): - if key in target: - return _coerce_bool(target.get(key), True) if key in runtime_target: return _coerce_bool(runtime_target.get(key), True) return True -def _target_service_names(target: dict[str, Any]) -> list[str]: - runtime_target = _runtime_target(target) - for key in ("service", "service_name", "cloud_run_service"): - value = target.get(key) or runtime_target.get(key) - if value: - return _split_values(str(value)) +def _target_service_names( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> list[str]: + defaults = defaults or {} + runtime_target = _runtime_target(target, defaults) + value = _target_field( + target, + defaults, + "service", + "service_name", + "cloud_run_service", + ) + if value is None: + for key in ("service", "service_name", "cloud_run_service"): + if runtime_target.get(key): + value = runtime_target[key] + break + if value: + return _split_values(str(value)) return [] def _region_for_service(service: str) -> str: - for target in _target_payloads(): - if service not in _target_service_names(target): + targets, defaults = _target_configuration() + for target in targets: + if service not in _target_service_names(target, defaults): continue - runtime_target = _runtime_target(target) + runtime_target = _runtime_target(target, defaults) for key in ("region", "cloud_run_region", "location"): - value = target.get(key) or runtime_target.get(key) + value = _target_field(target, defaults, key) or runtime_target.get(key) if value: return str(value).strip() return ( @@ -505,6 +607,9 @@ def main() -> int: issues: list[str] = [] details: list[str] = [] success_count = 0 + success_count_by_service: dict[str, int] = {} + queried_services: set[str] = set() + cloud_run_failures_by_service: dict[str, list[dict[str, Any]]] = {} service_since_by_name: dict[str, dt.datetime] = {} try: @@ -527,16 +632,26 @@ def main() -> int: except RuntimeError as exc: issues.append(f"Cloud Run log query failed for {service}: {exc}") continue + queried_services.add(service) failures = [entry for entry in entries if _is_failure(entry)] - success_count += sum(1 for entry in entries if _is_success(entry)) + cloud_run_failures_by_service[service] = failures + service_success_count = sum(1 for entry in entries if _is_success(entry)) + success_count_by_service[service] = service_success_count + success_count += service_success_count if failures: issues.append(f"{len(failures)} Cloud Run failure log(s) for {service}") details.extend(_summarize(entry) for entry in failures[:5]) - if services and require_success and success_count == 0: - issues.append( - f"no successful Cloud Run request found for {', '.join(services)} in the last {lookback_minutes} minutes" - ) + if services and require_success: + for service in _services_without_success( + services, + success_count_by_service, + queried_services, + ): + issues.append( + f"no successful Cloud Run request found for {service} " + f"in the last {lookback_minutes} minutes" + ) if check_scheduler and scheduler_pattern: log_filter = f'resource.type="cloud_scheduler_job" AND timestamp >= "{since_text}"' @@ -547,7 +662,7 @@ def main() -> int: entries = [ entry for entry in entries - if regex.search(str(_labels(entry).get("job_id") or _labels(entry).get("job_name") or "")) + if regex.search(_entry_job_name(entry).rsplit("/", 1)[-1]) ] failures = [] for entry in entries: @@ -557,6 +672,11 @@ def main() -> int: entry_since = _scheduler_entry_since(entry, service_since_by_name, since) if entry_timestamp and entry_timestamp < entry_since: continue + if _is_duplicate_scheduler_failure( + entry, + cloud_run_failures_by_service, + ): + continue failures.append(entry) if failures: issues.append(f"{len(failures)} Cloud Scheduler failure log(s)") diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index a327beb..92f4ec6 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -14,6 +14,31 @@ from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +try: + from scripts.runtime_heartbeat_policy import ( + filter_due_targets, + filter_services_for_targets, + load_runtime_targets, + match_payload_target, + runtime_target_configuration_has_enabled_targets, + runtime_target_configuration_present, + target_key, + target_label, + target_latest_due_at, + ) +except ModuleNotFoundError: + from runtime_heartbeat_policy import ( # type: ignore[no-redef] + filter_due_targets, + filter_services_for_targets, + load_runtime_targets, + match_payload_target, + runtime_target_configuration_has_enabled_targets, + runtime_target_configuration_present, + target_key, + target_label, + target_latest_due_at, + ) + DEFAULT_ACCEPT_STATUSES = {"ok", "skipped", "success", "completed", "no_action"} DEFAULT_REJECT_STATUSES = {"error", "failed", "failure", "cancelled", "canceled", "timed_out"} @@ -177,7 +202,7 @@ def _base_report_uris() -> list[str]: uris.extend(_split_values(os.environ.get("EXECUTION_REPORT_GCS_URI"))) state_bucket = (os.environ.get("FIRSTRADE_GCS_STATE_BUCKET") or "").strip() - if state_bucket: + if not uris and state_bucket: state_prefix = (os.environ.get("FIRSTRADE_STATE_PREFIX") or "firstrade-platform").strip("/") base = f"gs://{state_bucket}" if state_prefix: @@ -196,6 +221,8 @@ def _base_report_uris() -> list[str]: def _load_required_services() -> list[str]: services = [] + enabled_target_services = [] + disabled_target_services = [] for name in ( "RUNTIME_HEARTBEAT_REQUIRED_SERVICES", "CLOUD_RUN_SERVICES", @@ -218,20 +245,44 @@ def _load_required_services() -> list[str]: runtime_target = json.loads(runtime_target) except json.JSONDecodeError: runtime_target = {} + enabled_value = target.get("runtime_target_enabled") + if enabled_value is None: + enabled_value = target.get("RUNTIME_TARGET_ENABLED") + if enabled_value is None and isinstance(runtime_target, dict): + enabled_value = runtime_target.get("runtime_target_enabled") for key in ("service", "service_name", "cloud_run_service"): value = target.get(key) or ( runtime_target.get(key) if isinstance(runtime_target, dict) else None ) if value: - services.extend(_split_values(str(value))) + target_services = _split_values(str(value)) + if _enabled_value(enabled_value, default=True): + enabled_target_services.extend(target_services) + else: + disabled_target_services.extend(target_services) break except json.JSONDecodeError: pass + configured_targets = load_runtime_targets(os.environ, include_disabled=True) + if configured_targets: + enabled_target_services = [ + str(target.get("service") or "").strip() + for target in configured_targets + if target.get("enabled") is True + ] + disabled_target_services = [ + str(target.get("service") or "").strip() + for target in configured_targets + if target.get("enabled") is False + ] + + services.extend(enabled_target_services) + disabled = set(disabled_target_services) - set(enabled_target_services) seen = set() unique = [] for service in services: - if service not in seen: + if service not in seen and service not in disabled: seen.add(service) unique.append(service) return unique @@ -255,6 +306,93 @@ def _run_gcloud(args: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(args, text=True, capture_output=True, check=False) +def _scheduler_location() -> str: + return ( + os.environ.get("RUNTIME_HEARTBEAT_SCHEDULER_LOCATION") + or os.environ.get("CLOUD_RUN_REGION") + or "us-central1" + ) + + +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 list(dict.fromkeys(candidates)) + + +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 _hydrate_runtime_target_schedules( + targets: list[dict[str, Any]], + *, + project: str | None, +) -> list[dict[str, Any]]: + hydrated: list[dict[str, Any]] = [] + for target in targets: + scheduler = target.get("scheduler") + effective_scheduler = dict(scheduler) if isinstance(scheduler, dict) else {} + if len(str(effective_scheduler.get("main_time") or "").split()) == 5: + hydrated.append(target) + continue + if not _env_bool("RUNTIME_HEARTBEAT_SCHEDULER_AWARE", True): + raise RuntimeError( + "runtime target schedule is incomplete and scheduler lookup is disabled" + ) + service = str(target.get("service") or "").strip() + job = None + for job_name in _scheduler_job_name_candidates(service): + job = _describe_scheduler_job(job_name, project=project) + if job is not None: + break + schedule = str((job or {}).get("schedule") or "").strip() + if len(schedule.split()) != 5: + raise RuntimeError( + f"unable to resolve effective five-field scheduler cron for {service or ''}" + ) + timezone_name = str((job or {}).get("timeZone") or "").strip() + effective_scheduler["main_time"] = schedule + if timezone_name: + effective_scheduler["timezone"] = timezone_name + hydrated.append({**target, "scheduler": effective_scheduler}) + return hydrated + + def _list_gcs_objects(gcs_glob: str, *, project: str | None) -> list[dict[str, Any]]: command = ["gcloud", "storage", "ls", "--json", gcs_glob] if project: @@ -320,6 +458,20 @@ def _report_status(payload: dict[str, Any]) -> tuple[str, str]: return status, stage +def _report_notification_failure(payload: dict[str, Any]) -> str: + scopes = [payload] + summary = payload.get("summary") + if isinstance(summary, dict): + scopes.append(summary) + for scope in scopes: + delivery_summary = scope.get("notification_delivery_summary") + if isinstance(delivery_summary, dict) and delivery_summary.get("all_acknowledged") is False: + return "notification delivery not acknowledged" + if scope.get("notification_error") and scope.get("notification_suppressed") is not True: + return "notification delivery failed" + return "" + + def _payload_service_name(payload: dict[str, Any]) -> str: runtime_target = payload.get("runtime_target") service = payload.get("service_name") @@ -336,7 +488,12 @@ def _payload_account_scope(payload: dict[str, Any]) -> str: return "" -def _payload_matches(payload: dict[str, Any], required_services: list[str]) -> tuple[bool, str, str]: +def _payload_matches( + payload: dict[str, Any], + required_services: list[str], + *, + required_targets: list[dict[str, Any]] | None = None, +) -> tuple[bool, str, str]: expected_platform = (os.environ.get("RUNTIME_HEARTBEAT_REPORT_PLATFORM") or "").strip().lower() if expected_platform: platform = str(payload.get("platform") or "").strip().lower() @@ -352,6 +509,16 @@ def _payload_matches(payload: dict[str, Any], required_services: list[str]) -> t service_name = _payload_service_name(payload) if required_services and service_name not in required_services: return False, service_name, f"service_name={service_name or '-'}" + if required_targets: + matched_target, reason = match_payload_target(payload, required_targets) + if matched_target: + return True, matched_target, reason + target_services = { + str(target.get("service") or "").strip() + for target in required_targets + } + if service_name in target_services: + return False, service_name, reason return True, service_name, "matched filters" @@ -378,8 +545,11 @@ def _is_accepted_report(payload: dict[str, Any]) -> tuple[bool, str]: status_key = status.lower() stage_key = stage.upper() errors = _report_errors(payload) + notification_failure = _report_notification_failure(payload) if errors and not allow_errors: return False, f"errors={len(errors)} status={status or '-'} stage={stage or '-'}" + if notification_failure: + return False, notification_failure if status_key in reject_statuses or stage_key in reject_stages: return False, f"rejected status={status or '-'} stage={stage or '-'}" if status_key and status_key in accepted_statuses: @@ -465,6 +635,15 @@ def main(now: dt.datetime | None = None) -> int: if not _runtime_target_enabled(): print(f"Execution report heartbeat skipped for {name}: runtime target is disabled") return 0 + if ( + runtime_target_configuration_present(os.environ) + and not runtime_target_configuration_has_enabled_targets(os.environ) + ): + print( + f"Execution report heartbeat skipped for {name}: " + "no enabled runtime target matches this heartbeat" + ) + return 0 lookback_hours = float(os.environ.get("RUNTIME_HEARTBEAT_LOOKBACK_HOURS") or "36") max_reports = int(os.environ.get("RUNTIME_HEARTBEAT_MAX_REPORTS_TO_READ") or "20") fail_workflow = _env_bool("RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT", True) @@ -475,10 +654,77 @@ def main(now: dt.datetime | None = None) -> int: now = now.replace(tzinfo=dt.timezone.utc) now = now.astimezone(dt.timezone.utc) since = now - dt.timedelta(hours=lookback_hours) - schedule_skip_reason = _heartbeat_skip_reason_for_schedule(since, now) - if schedule_skip_reason: - print(f"Execution report heartbeat skipped for {name}: {schedule_skip_reason}") + runtime_targets = load_runtime_targets(os.environ) + try: + runtime_targets = _hydrate_runtime_target_schedules( + runtime_targets, + project=project, + ) + except RuntimeError as exc: + message = f"[Execution Report Heartbeat] {name}\nScheduler policy error: {exc}" + print(message) + _send_telegram(message) + return 1 if fail_workflow else 0 + publication_grace_minutes = float( + os.environ.get("RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES") or "30" + ) + if publication_grace_minutes < 0: + raise ValueError("RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES must be non-negative") + due_targets, target_schedule_evaluated = filter_due_targets( + runtime_targets, + since=since, + now=now, + market_aware=_env_bool("RUNTIME_HEARTBEAT_MARKET_AWARE", True), + publication_grace=dt.timedelta(minutes=publication_grace_minutes), + ) + if runtime_targets and target_schedule_evaluated and not due_targets: + target_names = ", ".join(target_label(target) for target in runtime_targets) + schedule_reason = _heartbeat_skip_reason_for_schedule(since, now) + print( + f"Execution report heartbeat skipped for {name}: " + + ( + schedule_reason + or "no runtime target main schedule was due on an open market session " + f"({target_names})" + ) + ) return 0 + if not runtime_targets: + schedule_skip_reason = _heartbeat_skip_reason_for_schedule(since, now) + if schedule_skip_reason: + print(f"Execution report heartbeat skipped for {name}: {schedule_skip_reason}") + return 0 + if due_targets: + required_services = filter_services_for_targets( + required_services, + due_targets, + all_targets=runtime_targets, + ) + required_targets = [ + target + for target in due_targets + if not required_services or str(target.get("service") or "").strip() in required_services + ] + required_keys = [target_key(target) for target in required_targets] + required_labels = { + target_key(target): target_label(target) + for target in required_targets + } + required_due_at = { + target_key(target): due_at + for target in required_targets + if (due_at := target_latest_due_at(target)) is not None + } + target_services = { + str(target.get("service") or "").strip() + for target in required_targets + } + for service in required_services: + if service not in target_services: + required_keys.append(service) + required_labels[service] = service + if required_keys: + max_reports = max(max_reports, min(200, len(required_keys) * 4)) globs = _report_globs(since, now) if not globs: @@ -505,23 +751,35 @@ def main(now: dt.datetime | None = None) -> int: if payload is None: inspected.append(f"- {updated.isoformat()} {uri} unreadable") continue - matches, service_name, filter_reason = _payload_matches(payload, required_services) + matches, service_name, filter_reason = _payload_matches( + payload, + required_services, + required_targets=required_targets, + ) if not matches: inspected.append(f"- {updated.isoformat()} {uri} skipped {filter_reason}") continue + latest_due_at = required_due_at.get(service_name) + if latest_due_at is not None and updated < latest_due_at: + inspected.append( + f"- {updated.isoformat()} {uri} skipped " + f"predates latest due schedule {latest_due_at.isoformat()}" + ) + continue ok, reason = _is_accepted_report(payload) inspected.append(f"- {updated.isoformat()} {uri} {reason}") if ok: - if required_services: + if required_keys: accepted_by_service[service_name] = (uri, updated, reason) else: accepted.append((uri, updated, reason)) - if required_services: - missing = [service for service in required_services if service not in accepted_by_service] + if required_keys: + missing = [key for key in required_keys if key not in accepted_by_service] if not missing: details = ", ".join( - f"{service}@{accepted_by_service[service][1].isoformat()}" for service in required_services + f"{required_labels[key]}@{accepted_by_service[key][1].isoformat()}" + for key in required_keys ) print(f"Execution report heartbeat OK for {name}: {details}") return 0 @@ -537,9 +795,12 @@ def main(now: dt.datetime | None = None) -> int: issues.extend(f"list failed: {item}" for item in list_errors[:3]) if not sorted_objects: issues.append(f"no report object updated in the last {lookback_hours:g} hours") - elif required_services: - missing = [service for service in required_services if service not in accepted_by_service] - issues.append(f"missing acceptable report for service(s): {', '.join(missing)}") + elif required_keys: + missing = [key for key in required_keys if key not in accepted_by_service] + issues.append( + "missing acceptable report for runtime target(s): " + + ", ".join(required_labels[key] for key in missing) + ) else: issues.append(f"no acceptable report among {min(len(sorted_objects), max_reports)} recent report object(s)") diff --git a/scripts/runtime_heartbeat_policy.py b/scripts/runtime_heartbeat_policy.py new file mode 100644 index 0000000..4baf3cb --- /dev/null +++ b/scripts/runtime_heartbeat_policy.py @@ -0,0 +1,677 @@ +"""Per-runtime-target schedule and market-session policy for heartbeat checks.""" + +from __future__ import annotations + +import datetime as dt +import json +import sys +from collections.abc import Callable, Mapping +from typing import Any +from zoneinfo import ZoneInfo + + +SessionDatesLoader = Callable[..., set[dt.date]] +WarningLogger = Callable[[str], None] +ProfileResolver = Callable[[str], str] + +_LATEST_DUE_AT_KEY = "_heartbeat_latest_due_at" +_MARKET_DEFAULTS = { + "US": ("NYSE", "America/New_York"), + "HK": ("XHKG", "Asia/Hong_Kong"), + "CN": ("SSE", "Asia/Shanghai"), + "SG": ("XSES", "Asia/Singapore"), + "CRYPTO": ("24/7", "UTC"), +} +_TIMEZONE_MARKETS = { + "America/New_York": "US", + "Asia/Hong_Kong": "HK", + "Asia/Shanghai": "CN", + "Asia/Singapore": "SG", +} +_PLATFORM_MARKETS = { + "schwab": "US", + "firstrade": "US", + "qmt": "CN", + "binance": "CRYPTO", +} + + +def _split_values(raw: str | None) -> list[str]: + if not raw: + return [] + values = str(raw).replace(";", ",").replace("\n", ",").split(",") + return [value.strip() for value in values if value.strip()] + + +def _enabled(value: Any, *, default: bool = True) -> bool: + if value is None or not str(value).strip(): + return default + return str(value).strip().lower() not in {"0", "false", "no", "n", "off"} + + +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _target_field( + item: Mapping[str, Any], + defaults: Mapping[str, Any], + *names: str, +) -> Any: + for source in ( + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ): + for name in names: + if name in source: + return source[name] + return None + + +def _runtime_target( + item: Mapping[str, Any], + defaults: Mapping[str, Any], +) -> dict[str, Any]: + value = _target_field(item, defaults, "runtime_target", "runtime_target_json") + if isinstance(value, str): + try: + value = json.loads(value) + except json.JSONDecodeError as exc: + raise ValueError(f"runtime target JSON is invalid: {exc}") from exc + if isinstance(value, Mapping): + return dict(value) + return dict(item) + + +def _first_value(sources: list[Mapping[str, Any]], keys: tuple[str, ...]) -> str: + for source in sources: + for key in keys: + value = source.get(key) + if value is not None and str(value).strip(): + return str(value).strip() + return "" + + +def _suffix_value(sources: list[Mapping[str, Any]], suffix: str) -> str: + for source in sources: + for key, value in source.items(): + if str(key).upper().endswith(suffix) and value is not None and str(value).strip(): + return str(value).strip() + return "" + + +def _service_values( + item: Mapping[str, Any], + runtime_target: Mapping[str, Any], + defaults: Mapping[str, Any], + environ: Mapping[str, str], +) -> list[str]: + value = _target_field( + item, + defaults, + "service", + "service_name", + "cloud_run_service", + ) + if value is None: + value = _first_value( + [runtime_target], + ("service", "service_name", "cloud_run_service"), + ) + if value: + return _split_values(str(value)) + services = _split_values(environ.get("CLOUD_RUN_SERVICES")) + services.extend(_split_values(environ.get("CLOUD_RUN_SERVICE"))) + return list(dict.fromkeys(services)) + + +def _normalize_target( + item: Mapping[str, Any], + runtime_target: Mapping[str, Any], + defaults: Mapping[str, Any], + service: str, + environ: Mapping[str, str], + *, + use_global_market_fallback: bool, + profile_resolver: ProfileResolver | None, +) -> dict[str, Any]: + sources = [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ] + scheduler = next( + ( + value + for source in sources + if isinstance((value := source.get("scheduler")), dict) + ), + {}, + ) + scheduler = dict(scheduler) + if not str(scheduler.get("main_time") or "").strip(): + main_time = _target_field( + item, + defaults, + "CLOUD_SCHEDULER_MAIN_TIME", + "cloud_scheduler_main_time", + ) + if main_time is None: + main_time = environ.get("CLOUD_SCHEDULER_MAIN_TIME") + if main_time is not None and str(main_time).strip(): + scheduler["main_time"] = str(main_time).strip() + account_scope = _first_value( + sources, + ( + "account_scope", + "account_group", + "account_region", + "ACCOUNT_GROUP", + "ACCOUNT_REGION", + ), + ) + target_market_timezone = ( + _first_value(sources, ("market_timezone", "MARKET_TIMEZONE")) + or _suffix_value(sources, "_MARKET_TIMEZONE") + ) + global_market_timezone = ( + str(environ.get("RUNTIME_HEARTBEAT_MARKET_TIMEZONE") or "").strip() + or _suffix_value([environ], "_MARKET_TIMEZONE") + ) + market_timezone = target_market_timezone or ( + global_market_timezone if use_global_market_fallback else "" + ) + market = ( + _first_value(sources, ("market", "MARKET")) + or _suffix_value(sources, "_MARKET") + ).upper() + if not market: + market = _TIMEZONE_MARKETS.get(target_market_timezone, "") + if market not in _MARKET_DEFAULTS: + market = _TIMEZONE_MARKETS.get(str(scheduler.get("timezone") or "").strip(), "") + if not market: + platform_id = _first_value(sources, ("platform_id",)).lower() + market = _PLATFORM_MARKETS.get(platform_id, "") + if not market and account_scope.upper() in {"US", "HK", "CN"}: + market = account_scope.upper() + if market not in _MARKET_DEFAULTS and use_global_market_fallback: + market = str(environ.get("RUNTIME_HEARTBEAT_MARKET") or "").strip().upper() + target_market_calendar = ( + _first_value(sources, ("market_calendar", "MARKET_CALENDAR")) + or _suffix_value(sources, "_MARKET_CALENDAR") + ) + global_market_calendar = ( + str(environ.get("RUNTIME_HEARTBEAT_MARKET_CALENDAR") or "").strip() + or _suffix_value([environ], "_MARKET_CALENDAR") + ) + default_calendar, default_timezone = _MARKET_DEFAULTS.get(market, ("", "")) + market_calendar = ( + target_market_calendar + or (global_market_calendar if use_global_market_fallback else "") + or default_calendar + ) + market_timezone = ( + market_timezone + or default_timezone + or str(scheduler.get("timezone") or "").strip() + ) + if scheduler and not str(scheduler.get("timezone") or "").strip() and market_timezone: + scheduler["timezone"] = market_timezone + strategy_profile = _first_value( + sources, + ("strategy_profile", "strategy", "profile"), + ) + if strategy_profile and profile_resolver is not None: + strategy_profile = profile_resolver(strategy_profile) + return { + "service": str(service).strip(), + "strategy_profile": strategy_profile, + "account_scope": account_scope, + "scheduler": scheduler, + "market": market, + "market_calendar": market_calendar, + "market_timezone": market_timezone, + } + + +def _load_runtime_target_items( + environ: Mapping[str, str], +) -> tuple[list[Mapping[str, Any]], Mapping[str, Any]]: + raw_targets = str(environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() + items: list[Mapping[str, Any]] = [] + defaults: Mapping[str, Any] = {} + if raw_targets: + try: + payload = json.loads(raw_targets) + except json.JSONDecodeError as exc: + raise ValueError(f"CLOUD_RUN_SERVICE_TARGETS_JSON is invalid: {exc}") from exc + if isinstance(payload, Mapping): + targets = payload.get("targets") + defaults = _mapping(payload.get("defaults")) + else: + targets = payload + if isinstance(targets, list): + items = [target for target in targets if isinstance(target, Mapping)] + else: + raise ValueError( + "CLOUD_RUN_SERVICE_TARGETS_JSON must be an array or object with targets" + ) + + if not items: + raw_runtime_target = str(environ.get("RUNTIME_TARGET_JSON") or "").strip() + if raw_runtime_target: + try: + runtime_target = json.loads(raw_runtime_target) + except json.JSONDecodeError as exc: + raise ValueError(f"RUNTIME_TARGET_JSON is invalid: {exc}") from exc + if isinstance(runtime_target, Mapping): + items = [runtime_target] + else: + raise ValueError("RUNTIME_TARGET_JSON must decode to an object") + return items, defaults + + +def runtime_target_configuration_present(environ: Mapping[str, str]) -> bool: + return bool( + str(environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() + or str(environ.get("RUNTIME_TARGET_JSON") or "").strip() + ) + + +def runtime_target_configuration_has_enabled_targets( + environ: Mapping[str, str], +) -> bool: + items, defaults = _load_runtime_target_items(environ) + if not items: + return False + expected_scope = str(environ.get("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE") or "").strip().lower() + for item in items: + runtime_target = _runtime_target(item, defaults) + target_scope = _first_value( + [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ], + ( + "account_scope", + "account_group", + "account_region", + "ACCOUNT_GROUP", + "ACCOUNT_REGION", + ), + ) + if expected_scope and target_scope and target_scope.lower() != expected_scope: + continue + enabled_value = _target_field( + item, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) + if enabled_value is None: + enabled_value = runtime_target.get("runtime_target_enabled") + if _enabled(enabled_value): + return True + return False + + +def load_runtime_targets( + environ: Mapping[str, str], + *, + include_disabled: bool = False, + profile_resolver: ProfileResolver | None = None, +) -> list[dict[str, Any]]: + items, defaults = _load_runtime_target_items(environ) + + expected_scope = str(environ.get("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE") or "").strip().lower() + eligible: list[tuple[Mapping[str, Any], dict[str, Any], bool]] = [] + for item in items: + runtime_target = _runtime_target(item, defaults) + enabled_value = _target_field( + item, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) + if enabled_value is None: + enabled_value = runtime_target.get("runtime_target_enabled") + enabled = _enabled(enabled_value) + if not enabled and not include_disabled: + continue + target_scope = _first_value( + [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ], + ( + "account_scope", + "account_group", + "account_region", + "ACCOUNT_GROUP", + "ACCOUNT_REGION", + ), + ) + if expected_scope and target_scope and target_scope.lower() != expected_scope: + continue + eligible.append((item, runtime_target, enabled)) + + normalized: list[dict[str, Any]] = [] + enabled_count = sum(1 for _item, _runtime_target_value, enabled in eligible if enabled) + for item, runtime_target, enabled in eligible: + for service in _service_values(item, runtime_target, defaults, environ): + target = _normalize_target( + item, + runtime_target, + defaults, + service, + environ, + use_global_market_fallback=enabled_count == 1, + profile_resolver=profile_resolver, + ) + if include_disabled: + target["enabled"] = enabled + key = target_key(target) + if key and all(target_key(existing) != key for existing in normalized): + normalized.append(target) + return normalized + + +def target_key(target: Mapping[str, Any]) -> str: + service = str(target.get("service") or "").strip().lower() + strategy = str(target.get("strategy_profile") or "").strip().lower() + scope = str(target.get("account_scope") or "").strip().lower() + return f"{service}|{strategy or '*'}|{scope or '*'}" if service else "" + + +def target_label(target: Mapping[str, Any]) -> str: + service = str(target.get("service") or "").strip() or "" + strategy = str(target.get("strategy_profile") or "").strip() + scope = str(target.get("account_scope") or "").strip() + qualifiers = "/".join(value for value in (strategy, scope) if value) + return f"{service}[{qualifiers}]" if qualifiers else service + + +def target_latest_due_at(target: Mapping[str, Any]) -> dt.datetime | None: + value = target.get(_LATEST_DUE_AT_KEY) + return value if isinstance(value, dt.datetime) else None + + +def _payload_value(payload: Mapping[str, Any], keys: tuple[str, ...]) -> str: + runtime_target = payload.get("runtime_target") + sources = [payload] + if isinstance(runtime_target, Mapping): + sources.append(runtime_target) + return _first_value(sources, keys) + + +def match_payload_target( + payload: Mapping[str, Any], + targets: list[dict[str, Any]], +) -> tuple[str | None, str]: + service = _payload_value( + payload, + ("service_name", "service", "cloud_run_service"), + ).lower() + strategy = _payload_value( + payload, + ("strategy_profile", "strategy", "profile"), + ).lower() + scope = _payload_value( + payload, + ("account_scope", "account_group", "account_region"), + ).lower() + for target in targets: + expected_service = str(target.get("service") or "").strip().lower() + expected_strategy = str(target.get("strategy_profile") or "").strip().lower() + expected_scope = str(target.get("account_scope") or "").strip().lower() + if service != expected_service: + continue + if expected_strategy and strategy != expected_strategy: + continue + if expected_scope and scope != expected_scope: + continue + return target_key(target), "matched runtime target" + return None, ( + f"runtime_target={service or '-'}/{strategy or '-'}/{scope or '-'}" + ) + + +def _cron_token_value(token: str, *, names: dict[str, int] | None = None) -> int: + normalized = token.strip().lower() + if names and normalized in names: + return names[normalized] + return int(normalized) + + +def _cron_field_values( + field: str, + *, + minimum: int, + maximum: int, + names: dict[str, int] | None = None, +) -> set[int] | None: + text = str(field or "").strip().lower() + if text in {"", "*"}: + return None + values: set[int] = set() + for raw_part in text.split(","): + part = raw_part.strip() + if not part: + continue + base, raw_step = part, "1" + if "/" in part: + base, raw_step = part.split("/", 1) + step = max(1, int(raw_step)) + if base == "*": + start, end = minimum, maximum + elif "-" in base: + raw_start, raw_end = base.split("-", 1) + start = _cron_token_value(raw_start, names=names) + end = _cron_token_value(raw_end, names=names) + else: + start = end = _cron_token_value(base, names=names) + for value in range(start, end + 1, step): + if minimum <= value <= maximum: + values.add(value) + elif maximum == 6 and value == 7: + values.add(0) + return values + + +def cron_matches(schedule: str, value: dt.datetime) -> bool: + fields = str(schedule or "").split() + if len(fields) == 2: + fields.extend(("*", "*", "*")) + if len(fields) != 5: + return False + minute, hour, day_of_month, month, day_of_week = fields + dow_names = { + "sun": 0, + "mon": 1, + "tue": 2, + "wed": 3, + "thu": 4, + "fri": 5, + "sat": 6, + } + minute_values = _cron_field_values(minute, minimum=0, maximum=59) + hour_values = _cron_field_values(hour, minimum=0, maximum=23) + dom_values = _cron_field_values(day_of_month, minimum=1, maximum=31) + month_values = _cron_field_values(month, minimum=1, maximum=12) + dow_values = _cron_field_values(day_of_week, minimum=0, maximum=6, names=dow_names) + if minute_values is not None and value.minute not in minute_values: + return False + if hour_values is not None and value.hour not in hour_values: + return False + if month_values is not None and value.month not in month_values: + return False + dom_matches = dom_values is None or value.day in dom_values + dow_matches = dow_values is None or value.isoweekday() % 7 in dow_values + if dom_values is not None and dow_values is not None: + return dom_matches or dow_matches + return dom_matches and dow_matches + + +def _market_session_dates( + calendar: str, + *, + start_date: dt.date, + end_date: dt.date, +) -> set[dt.date]: + import pandas_market_calendars as mcal + + schedule = mcal.get_calendar(calendar).schedule( + start_date=start_date, + end_date=end_date, + ) + return {value.date() for value in schedule.index} + + +def _target_due_status( + target: Mapping[str, Any], + *, + since: dt.datetime, + now: dt.datetime, + market_aware: bool, + session_dates_loader: SessionDatesLoader, + warning_logger: WarningLogger, + publication_grace: dt.timedelta, +) -> tuple[bool | None, dt.datetime | None]: + scheduler = target.get("scheduler") + if not isinstance(scheduler, Mapping): + return None, None + schedule = str(scheduler.get("main_time") or "").strip() + fields = schedule.split() + if len(fields) != 5: + return None, None + timezone_name = str(scheduler.get("timezone") or "UTC").strip() or "UTC" + try: + scheduler_timezone = ZoneInfo(timezone_name) + except Exception as exc: # noqa: BLE001 + warning_logger( + f"Unable to evaluate heartbeat scheduler timezone {timezone_name}: " + f"{type(exc).__name__}; keeping target required" + ) + return None, None + + since_utc = since.astimezone(dt.timezone.utc) + now_utc = now.astimezone(dt.timezone.utc) + cursor = since_utc.replace(second=0, microsecond=0) + if cursor < since_utc: + cursor += dt.timedelta(minutes=1) + matured_at = now_utc - max(publication_grace, dt.timedelta()) + cron_due_at: list[dt.datetime] = [] + while cursor <= now_utc: + local_time = cursor.astimezone(scheduler_timezone) + try: + matches = cron_matches(schedule, local_time) + except (TypeError, ValueError) as exc: + warning_logger( + f"Unable to evaluate heartbeat cron for {target_label(target)}: " + f"{type(exc).__name__}; keeping target required" + ) + return None, None + if matches and cursor <= matured_at: + cron_due_at.append(cursor) + cursor += dt.timedelta(minutes=1) + latest_cron_due_at = cron_due_at[-1] if cron_due_at else None + + session_dates: set[dt.date] | None = None + market_calendar = str(target.get("market_calendar") or "").strip() + market_timezone_name = ( + str(target.get("market_timezone") or "").strip() or timezone_name + ) + if market_aware and market_calendar: + try: + market_timezone = ZoneInfo(market_timezone_name) + session_dates = session_dates_loader( + market_calendar, + start_date=since_utc.astimezone(market_timezone).date(), + end_date=now_utc.astimezone(market_timezone).date(), + ) + except Exception as exc: # noqa: BLE001 + warning_logger( + f"Unable to evaluate heartbeat market calendar {market_calendar}: " + f"{type(exc).__name__}; keeping target required" + ) + return None, latest_cron_due_at + + latest_due_at = latest_cron_due_at + if session_dates is not None: + market_timezone = ZoneInfo(market_timezone_name) + latest_due_at = next( + ( + due_at + for due_at in reversed(cron_due_at) + if due_at.astimezone(market_timezone).date() in session_dates + ), + None, + ) + return latest_due_at is not None, latest_due_at + + +def filter_due_targets( + targets: list[dict[str, Any]], + *, + since: dt.datetime, + now: dt.datetime, + market_aware: bool = True, + publication_grace: dt.timedelta = dt.timedelta(minutes=30), + session_dates_loader: SessionDatesLoader = _market_session_dates, + warning_logger: WarningLogger = lambda message: print(message, file=sys.stderr), +) -> tuple[list[dict[str, Any]], bool]: + due: list[dict[str, Any]] = [] + evaluated = False + for target in targets: + status, latest_due_at = _target_due_status( + target, + since=since, + now=now, + market_aware=market_aware, + session_dates_loader=session_dates_loader, + warning_logger=warning_logger, + publication_grace=publication_grace, + ) + if status is not None: + evaluated = True + if status is not False: + due_target = dict(target) + if latest_due_at is not None: + due_target[_LATEST_DUE_AT_KEY] = latest_due_at + due.append(due_target) + return due, evaluated + + +def filter_services_for_targets( + services: list[str], + targets: list[dict[str, Any]], + *, + all_targets: list[dict[str, Any]] | None = None, +) -> list[str]: + if not targets: + return services + target_services = { + str(target.get("service") or "").strip() + for target in targets + if str(target.get("service") or "").strip() + } + configured_services = { + str(target.get("service") or "").strip() + for target in (all_targets or targets) + if str(target.get("service") or "").strip() + } + return [ + service + for service in services + if service not in configured_services or service in target_services + ] diff --git a/tests/test_cloud_run_runtime_guard.py b/tests/test_cloud_run_runtime_guard.py index 861c700..54227fa 100644 --- a/tests/test_cloud_run_runtime_guard.py +++ b/tests/test_cloud_run_runtime_guard.py @@ -139,6 +139,54 @@ def test_scheduler_entry_since_uses_matching_service_revision_window(): ) +def test_scheduler_failure_matching_cloud_run_failure_is_duplicate(): + scheduler_entry = { + "timestamp": "2026-07-29T19:45:03Z", + "resource": {"labels": {"job_id": "firstrade-scheduler"}}, + } + cloud_run_failures = { + "firstrade-service": [ + { + "timestamp": "2026-07-29T19:45:01Z", + "resource": {"labels": {"service_name": "firstrade-service"}}, + } + ] + } + + assert guard._is_duplicate_scheduler_failure( + scheduler_entry, + cloud_run_failures, + ) + + +def test_scheduler_failure_for_other_service_is_not_duplicate(): + scheduler_entry = { + "timestamp": "2026-07-29T19:45:03Z", + "resource": {"labels": {"job_id": "other-platform-scheduler"}}, + } + cloud_run_failures = { + "firstrade-service": [ + { + "timestamp": "2026-07-29T19:45:01Z", + "resource": {"labels": {"service_name": "firstrade-service"}}, + } + ] + } + + assert not guard._is_duplicate_scheduler_failure( + scheduler_entry, + cloud_run_failures, + ) + + +def test_services_without_success_are_reported_individually(): + assert guard._services_without_success( + ["healthy-service", "silent-service"], + {"healthy-service": 1, "silent-service": 0}, + {"healthy-service", "silent-service"}, + ) == ["silent-service"] + + def test_monitor_dispatch_capacity_warning_is_not_failure_by_default(monkeypatch): monkeypatch.delenv("RUNTIME_GUARD_IGNORE_MONITOR_DISPATCH_CAPACITY_WARNINGS", raising=False) entry = { @@ -179,3 +227,49 @@ def test_strategy_request_capacity_warning_still_fails(monkeypatch): } assert guard._is_failure(entry) is True + + +def test_scheduler_job_matching_rejects_prefixed_service_names(): + pattern = guard._scheduler_job_pattern_for_services(["test-service"]) + entry = { + "timestamp": "2026-07-29T19:45:03Z", + "resource": { + "labels": {"job_id": "test-secondary-service-scheduler"} + }, + } + failures = { + "test-service": [{"timestamp": "2026-07-29T19:45:01Z"}], + } + fallback = dt.datetime(2026, 7, 29, 19, 0, tzinfo=dt.timezone.utc) + service_since = dt.datetime(2026, 7, 29, 19, 30, tzinfo=dt.timezone.utc) + + assert not re.search(pattern, "test-secondary-service-scheduler") + assert guard._scheduler_entry_since( + entry, + {"test-service": service_since}, + fallback, + ) == fallback + assert guard._is_duplicate_scheduler_failure(entry, failures) is False + + +def test_explicit_disabled_service_is_removed_using_target_defaults(monkeypatch): + monkeypatch.delenv("RUNTIME_GUARD_CLOUD_RUN_SERVICES", raising=False) + monkeypatch.delenv("CLOUD_RUN_SERVICE", raising=False) + monkeypatch.setenv("CLOUD_RUN_SERVICES", "enabled-service,disabled-service") + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [ + { + "service": "enabled-service", + "runtime_target_enabled": True, + }, + {"service": "disabled-service"}, + ], + } + ), + ) + + assert guard._load_services() == ["enabled-service"] diff --git a/tests/test_execution_report_heartbeat.py b/tests/test_execution_report_heartbeat.py index 437f550..008eb30 100644 --- a/tests/test_execution_report_heartbeat.py +++ b/tests/test_execution_report_heartbeat.py @@ -2,10 +2,34 @@ import subprocess import datetime as dt +import json from scripts import execution_report_heartbeat as heartbeat +def test_required_services_skip_disabled_runtime_targets(monkeypatch): + for name in ( + "RUNTIME_HEARTBEAT_REQUIRED_SERVICES", + "CLOUD_RUN_SERVICES", + "CLOUD_RUN_SERVICE", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "defaults": {"RUNTIME_TARGET_ENABLED": "false"}, + "targets": [ + {"service": "firstrade-enabled-service", "RUNTIME_TARGET_ENABLED": "true"}, + {"service": "firstrade-disabled-service"}, + ] + } + ), + ) + + assert heartbeat._load_required_services() == ["firstrade-enabled-service"] + + def test_report_globs_include_sanitized_month_segments(monkeypatch): monkeypatch.delenv("RUNTIME_HEARTBEAT_GCS_GLOBS", raising=False) monkeypatch.delenv("EXECUTION_REPORT_GCS_URI", raising=False) @@ -26,6 +50,21 @@ def test_report_globs_include_sanitized_month_segments(monkeypatch): "gs://runtime-state/firstrade-platform/strategy-runs/**/2026_06/*.json", ] + +def test_execution_report_uri_disables_strategy_state_fallback(monkeypatch): + monkeypatch.delenv("RUNTIME_HEARTBEAT_GCS_URIS", raising=False) + monkeypatch.setenv( + "EXECUTION_REPORT_GCS_URI", + "gs://runtime-reports/execution-reports", + ) + monkeypatch.setenv("FIRSTRADE_GCS_STATE_BUCKET", "runtime-state") + monkeypatch.setenv("FIRSTRADE_STATE_PREFIX", "firstrade-platform") + + assert heartbeat._base_report_uris() == [ + "gs://runtime-reports/execution-reports", + ] + + def test_telegram_token_falls_back_to_secret_manager(monkeypatch): monkeypatch.delenv("TELEGRAM_TOKEN", raising=False) monkeypatch.delenv("TG_TOKEN", raising=False) @@ -88,6 +127,70 @@ def test_heartbeat_skips_when_runtime_target_json_is_disabled(monkeypatch, capsy assert "runtime target is disabled" in output +def test_heartbeat_skips_when_all_configured_targets_are_disabled( + monkeypatch, + capsys, +): + monkeypatch.delenv("RUNTIME_TARGET_ENABLED", raising=False) + monkeypatch.delenv("RUNTIME_TARGET_JSON", raising=False) + monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "Firstrade disabled targets") + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [{"service": "disabled-service"}], + } + ), + ) + monkeypatch.setattr( + heartbeat, + "_list_gcs_objects", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("GCS should not be queried") + ), + ) + + result = heartbeat.main( + now=dt.datetime(2026, 6, 20, 23, 10, tzinfo=dt.timezone.utc) + ) + + assert result == 0 + assert "no enabled runtime target matches this heartbeat" in capsys.readouterr().out + + +def test_incomplete_target_schedule_uses_deployed_scheduler_cron(monkeypatch): + targets = [ + { + "service": "firstrade-service", + "scheduler": { + "main_time": "45 15", + "timezone": "America/New_York", + }, + } + ] + monkeypatch.setattr( + heartbeat, + "_describe_scheduler_job", + lambda job_name, **_kwargs: ( + { + "name": job_name, + "schedule": "45 15 25-29 * *", + "timeZone": "America/New_York", + } + if job_name == "firstrade-service-scheduler" + else None + ), + ) + + hydrated = heartbeat._hydrate_runtime_target_schedules( + targets, + project="test-project", + ) + + assert hydrated[0]["scheduler"]["main_time"] == "45 15 25-29 * *" + + def test_heartbeat_skips_outside_runtime_target_scheduler_day(monkeypatch, capsys): monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "Firstrade monthly runtime") @@ -136,3 +239,19 @@ def test_heartbeat_does_not_skip_when_lookback_includes_scheduler_day(monkeypatc ) assert reason is None + + +def test_report_with_failed_notification_delivery_is_rejected(): + accepted, reason = heartbeat._is_accepted_report( + { + "status": "ok", + "summary": { + "notification_sent": False, + "notification_suppressed": False, + "notification_error": "delivery_not_acknowledged", + }, + } + ) + + assert accepted is False + assert "notification delivery failed" in reason diff --git a/tests/test_notifications.py b/tests/test_notifications.py index e5b99f2..ecf64cf 100644 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -13,18 +13,29 @@ class FakeRequests: - def __init__(self): + def __init__(self, *, status_code=200, payload=None): self.calls = [] + self.status_code = status_code + self.payload = {"ok": True} if payload is None else payload def post(self, url, json, timeout): self.calls.append((url, json, timeout)) - return object() + return FakeResponse(self.status_code, self.payload) + + +class FakeResponse: + def __init__(self, status_code, payload): + self.status_code = status_code + self._payload = payload + + def json(self): + return self._payload def test_build_sender_breaks_market_symbol_auto_links(): fake_requests = FakeRequests() sender = build_sender("token-1", "chat-1", requests_module=fake_requests) - sender("SOXL.US and 00700.HK") + assert sender("SOXL.US and 00700.HK") is True assert len(fake_requests.calls) == 1 url, payload, timeout = fake_requests.calls[0] @@ -32,3 +43,10 @@ def test_build_sender_breaks_market_symbol_auto_links(): assert payload["chat_id"] == "chat-1" assert payload["text"] == "SOXL.\u2060US and 00700.\u2060HK" assert timeout == 15 + + +def test_build_sender_returns_false_when_telegram_rejects_message(): + fake_requests = FakeRequests(payload={"ok": False, "description": "chat not found"}) + sender = build_sender("token-1", "chat-1", requests_module=fake_requests) + + assert sender("rebalance") is False diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index a603ed9..7dced56 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -5,7 +5,11 @@ from types import SimpleNamespace from application.firstrade_client import FirstradeCredentials -from application.rebalance_service import _runtime_metadata_with_execution_policy, run_strategy_cycle +from application.rebalance_service import ( + _publish_cycle_notification, + _runtime_metadata_with_execution_policy, + run_strategy_cycle, +) from notifications.telegram import I18N, build_translator, render_cycle_summary from quant_platform_kit.strategy_contracts import PositionTarget, StrategyDecision from runtime_config_support import PlatformRuntimeSettings @@ -37,6 +41,17 @@ def _runtime_settings_with_persistence(**overrides) -> PlatformRuntimeSettings: return PlatformRuntimeSettings(**values) +def test_publish_cycle_notification_propagates_delivery_failure(): + sent = _publish_cycle_notification( + {"ok": False, "error": "execution failed"}, + settings=_runtime_settings(), + notification_sender=lambda _message: False, + log_message=lambda _message: None, + ) + + assert sent is False + + def test_runtime_metadata_uses_platform_execution_policy_over_strategy_metadata(): metadata = { "signal": "ok", diff --git a/tests/test_runtime_heartbeat_policy.py b/tests/test_runtime_heartbeat_policy.py new file mode 100644 index 0000000..6e14a88 --- /dev/null +++ b/tests/test_runtime_heartbeat_policy.py @@ -0,0 +1,496 @@ +from __future__ import annotations + +import datetime as dt +import json + +from scripts.runtime_heartbeat_policy import ( + filter_due_targets, + load_runtime_targets, + match_payload_target, + runtime_target_configuration_present, + target_key, + target_latest_due_at, +) +from scripts import execution_report_heartbeat as heartbeat + + +def _target( + *, + service: str, + strategy: str, + scope: str, + timezone: str, + calendar: str, +) -> dict[str, object]: + return { + "service": service, + "runtime_target": { + "service_name": service, + "strategy_profile": strategy, + "account_scope": scope, + "scheduler": { + "timezone": timezone, + "main_time": "45 15 * * *", + }, + "market_calendar": calendar, + "market_timezone": timezone, + }, + } + + +def test_due_targets_use_each_strategy_market_calendar() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="svc-hk", + strategy="hk-strategy", + scope="HK", + timezone="Asia/Hong_Kong", + calendar="XHKG", + ), + ] + } + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 3, 0, 0, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 3, 22, 0, tzinfo=dt.timezone.utc), + session_dates_loader=lambda calendar, **_kwargs: ( + {dt.date(2026, 7, 3)} if calendar == "XHKG" else set() + ), + ) + + assert evaluated is True + assert [target["strategy_profile"] for target in due] == ["hk-strategy"] + + +def test_real_exchange_calendars_distinguish_us_holiday_from_hk_session() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="svc-hk", + strategy="hk-strategy", + scope="HK", + timezone="Asia/Hong_Kong", + calendar="XHKG", + ), + ] + } + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 3, 0, 0, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 3, 22, 0, tzinfo=dt.timezone.utc), + ) + + assert evaluated is True + assert [target["strategy_profile"] for target in due] == ["hk-strategy"] + + +def test_july_29_us_month_end_target_is_due_at_1545_eastern() -> None: + raw_target = _target( + service="svc-us-monthly", + strategy="us-monthly", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ) + raw_target["runtime_target"]["scheduler"]["main_time"] = "45 15 25-29 * *" + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + {"targets": [raw_target]} + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 29, 19, 40, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 29, 20, 20, tzinfo=dt.timezone.utc), + ) + + assert evaluated is True + assert [target["strategy_profile"] for target in due] == ["us-monthly"] + assert target_latest_due_at(due[0]) == dt.datetime( + 2026, + 7, + 29, + 19, + 45, + tzinfo=dt.timezone.utc, + ) + + +def test_neutral_daily_heartbeat_tracks_latest_due_time_per_market() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="svc-hk", + strategy="hk-strategy", + scope="HK", + timezone="Asia/Hong_Kong", + calendar="XHKG", + ), + ] + } + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 28, 10, 20, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 29, 22, 20, tzinfo=dt.timezone.utc), + session_dates_loader=lambda _calendar, **_kwargs: { + dt.date(2026, 7, 28), + dt.date(2026, 7, 29), + }, + ) + + assert evaluated is True + assert { + target["strategy_profile"]: target_latest_due_at(target) + for target in due + } == { + "us-strategy": dt.datetime( + 2026, + 7, + 29, + 19, + 45, + tzinfo=dt.timezone.utc, + ), + "hk-strategy": dt.datetime( + 2026, + 7, + 29, + 7, + 45, + tzinfo=dt.timezone.utc, + ), + } + + +def test_same_service_strategies_require_distinct_reports() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="shared-service", + strategy="strategy-a", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="shared-service", + strategy="strategy-b", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + ] + } + ) + } + ) + + matched, reason = match_payload_target( + { + "service_name": "shared-service", + "strategy_profile": "strategy-a", + "account_scope": "US", + }, + targets, + ) + + assert matched == target_key(targets[0]) + assert reason == "matched runtime target" + missing_strategy, _ = match_payload_target( + {"service_name": "shared-service", "account_scope": "US"}, + targets, + ) + assert missing_strategy is None + + matched_by_heartbeat, matched_key, _ = heartbeat._payload_matches( + { + "service_name": "shared-service", + "strategy_profile": "strategy-a", + "account_scope": "US", + }, + ["shared-service"], + required_targets=targets, + ) + assert matched_by_heartbeat is True + assert matched_key == target_key(targets[0]) + missing_by_heartbeat, _, _ = heartbeat._payload_matches( + {"service_name": "shared-service", "account_scope": "US"}, + ["shared-service"], + required_targets=targets, + ) + assert missing_by_heartbeat is False + + +def test_scheduler_timezone_beats_account_region_when_market_is_not_explicit() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + { + "service": "longbridge-sg-us-service", + "account_scope": "SG", + "runtime_target": { + "service_name": "longbridge-sg-us-service", + "strategy_profile": "us-strategy", + "account_scope": "SG", + "scheduler": { + "timezone": "America/New_York", + "main_time": "45 15 * * *", + }, + }, + } + ] + } + ) + } + ) + + assert targets[0]["market"] == "US" + assert targets[0]["market_calendar"] == "NYSE" + assert targets[0]["market_timezone"] == "America/New_York" + + +def test_ambiguous_sg_account_does_not_guess_a_stock_exchange() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + { + "service": "longbridge-sg-service", + "account_scope": "SG", + "runtime_target": { + "service_name": "longbridge-sg-service", + "strategy_profile": "unknown-strategy", + "account_scope": "SG", + "scheduler": { + "timezone": "UTC", + "main_time": "0 12 * * *", + }, + }, + } + ] + } + ) + } + ) + + assert targets[0]["market"] == "" + assert targets[0]["market_calendar"] == "" + + +def test_calendar_failure_keeps_target_due_fail_closed() -> None: + targets = load_runtime_targets( + { + "RUNTIME_TARGET_JSON": json.dumps( + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="INVALID", + )["runtime_target"] + ) + } + ) + + def fail_calendar(_calendar: str, **_kwargs: object) -> set[dt.date]: + raise RuntimeError("calendar unavailable") + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 3, 0, 0, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 3, 22, 0, tzinfo=dt.timezone.utc), + session_dates_loader=fail_calendar, + warning_logger=lambda _message: None, + ) + + assert len(due) == 1 + assert target_latest_due_at(due[0]) == dt.datetime( + 2026, + 7, + 3, + 19, + 45, + tzinfo=dt.timezone.utc, + ) + assert evaluated is False + + +def test_target_defaults_and_scheduler_aliases_are_normalized() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "defaults": { + "env": { + "RUNTIME_TARGET_ENABLED": "false", + "CLOUD_SCHEDULER_MAIN_TIME": "45 15 25-29 * *", + }, + "market": "US", + }, + "targets": [ + { + "service": "disabled-service", + "runtime_target": { + "strategy_profile": "disabled-strategy", + }, + }, + { + "service": "enabled-service", + "RUNTIME_TARGET_ENABLED": "true", + "runtime_target": { + "strategy_profile": "enabled-strategy", + }, + }, + ], + } + ) + } + ) + + assert len(targets) == 1 + assert targets[0]["service"] == "enabled-service" + assert targets[0]["market"] == "US" + assert targets[0]["scheduler"] == { + "main_time": "45 15 25-29 * *", + "timezone": "America/New_York", + } + + +def test_strategy_profile_resolver_canonicalizes_aliases() -> None: + targets = load_runtime_targets( + { + "RUNTIME_TARGET_JSON": json.dumps( + { + "service_name": "alias-service", + "strategy_profile": "supported-alias", + "scheduler": { + "main_time": "45 15 * * *", + "timezone": "UTC", + }, + } + ) + }, + profile_resolver=lambda value: ( + "canonical-strategy" if value == "supported-alias" else value + ), + ) + + assert targets[0]["strategy_profile"] == "canonical-strategy" + + +def test_publication_grace_uses_previous_matured_schedule_cutoff() -> None: + targets = load_runtime_targets( + { + "RUNTIME_TARGET_JSON": json.dumps( + { + "service_name": "grace-service", + "strategy_profile": "grace-strategy", + "scheduler": { + "main_time": "0 12 * * *", + "timezone": "UTC", + }, + } + ) + } + ) + since = dt.datetime(2026, 7, 28, 11, 30, tzinfo=dt.timezone.utc) + + within_grace, evaluated = filter_due_targets( + targets, + since=since, + now=dt.datetime(2026, 7, 29, 12, 5, tzinfo=dt.timezone.utc), + market_aware=False, + publication_grace=dt.timedelta(minutes=30), + ) + after_grace, _ = filter_due_targets( + targets, + since=since, + now=dt.datetime(2026, 7, 29, 12, 31, tzinfo=dt.timezone.utc), + market_aware=False, + publication_grace=dt.timedelta(minutes=30), + ) + + assert evaluated is True + assert target_latest_due_at(within_grace[0]) == dt.datetime( + 2026, + 7, + 28, + 12, + 0, + tzinfo=dt.timezone.utc, + ) + assert target_latest_due_at(after_grace[0]) == dt.datetime( + 2026, + 7, + 29, + 12, + 0, + tzinfo=dt.timezone.utc, + ) + + +def test_runtime_target_configuration_presence_is_preserved_when_all_disabled() -> None: + environ = { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [{"service": "disabled-service"}], + } + ) + } + + assert runtime_target_configuration_present(environ) is True + assert load_runtime_targets(environ) == [] + assert load_runtime_targets(environ, include_disabled=True)[0]["enabled"] is False diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py new file mode 100644 index 0000000..9e47878 --- /dev/null +++ b/tests/test_runtime_monitor_workflows.py @@ -0,0 +1,27 @@ +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_execution_report_heartbeat_has_market_neutral_daily_schedule() -> None: + workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text() + + assert 'cron: "20 22 * * *"' in workflow + assert 'cron: "20 22 * * 1-5"' not in workflow + assert "RUNTIME_HEARTBEAT_MARKET_AWARE:" in workflow + assert "RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES:" in workflow + assert "RUNTIME_HEARTBEAT_SCHEDULER_LOCATION:" in workflow + assert "CLOUD_SCHEDULER_MAIN_TIME:" in workflow + assert "EXECUTION_REPORT_GCS_URI:" in workflow + assert "pandas-market-calendars==5.4.0" in workflow + + +def test_runtime_monitor_workflows_retry_gcp_authentication() -> None: + for name in ("execution-report-heartbeat.yml", "runtime-guard.yml"): + workflow = (ROOT / ".github/workflows" / name).read_text() + + assert workflow.count("google-github-actions/auth@v3") == 2 + assert "id: gcp_auth_primary" in workflow + assert "continue-on-error: true" in workflow + assert "steps.gcp_auth_primary.outcome == 'failure'" in workflow diff --git a/tests/test_sync_cloud_run_env_workflow.py b/tests/test_sync_cloud_run_env_workflow.py index b4ae6bd..a154b56 100644 --- a/tests/test_sync_cloud_run_env_workflow.py +++ b/tests/test_sync_cloud_run_env_workflow.py @@ -14,6 +14,10 @@ def test_sync_cloud_run_env_workflow_uses_sync_plan_script(): assert "Cloud Run env sync did not resolve any targets" in workflow assert "Cloud Run sync target is missing service_name" in workflow assert "Cloud Run sync target {service_name} is missing env" in workflow + assert ( + "CLOUD_RUN_SERVICE_TARGETS_JSON: " + "${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }}" + ) in workflow for name in ( "CLOUD_SCHEDULER_LOCATION",