fix: harden runtime scheduling and notifications

.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: longbridgequant
@@ -56,8 +57,12 @@ jobs:
       RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT: ${{ inputs.fail_workflow_on_alert || vars.RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT || 'true' }}
       RUNTIME_HEARTBEAT_ACCEPT_STATUSES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STATUSES }}
       RUNTIME_HEARTBEAT_REJECT_STATUSES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STATUSES }}
+      RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }}
+      RUNTIME_HEARTBEAT_MARKET_CALENDAR: ${{ vars.LONGBRIDGE_MARKET_CALENDAR }}
+      RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.LONGBRIDGE_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 }}
+      RUNTIME_HEARTBEAT_SCHEDULER_LOCATION: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_LOCATION || vars.CLOUD_RUN_REGION || 'us-central1' }}
       RUNTIME_HEARTBEAT_EXPECTED_DAY_OF_MONTH: ${{ vars.RUNTIME_HEARTBEAT_EXPECTED_DAY_OF_MONTH }}
       RUNTIME_HEARTBEAT_EXPECTED_TIMEZONE: ${{ vars.RUNTIME_HEARTBEAT_EXPECTED_TIMEZONE }}
       RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }}
@@ -66,6 +71,7 @@ jobs:
       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 }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
       TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}
@@ -74,6 +80,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 }}
@@ -82,5 +101,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

.github/workflows/runtime-guard.yml

@@ -74,6 +74,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 }}

application/runtime_composer.py

@@ -100,7 +100,7 @@ def __post_init__(self) -> None:
     def with_prefix(self, message: str) -> str:
         return self.prefixer_builder(self.account_prefix)(message)
 
-    def send_message(self, message: str) -> None:
+    def send_message(self, message: str) -> bool:
         """Send a cycle notification through the configured channel."""
         prefixed = self.with_prefix(message)
         sender = build_cycle_sender(
@@ -109,7 +109,7 @@ def send_message(self, message: str) -> None:
             telegram_chat_id=self.tg_chat_id,
             webhook_url=self.webhook_url,
         )
-        sender(prefixed)
+        return sender(prefixed)
 
     send_tg_message = send_message  # backward-compat alias
 

application/runtime_notification_adapters.py

@@ -27,13 +27,18 @@ class LongBridgeNotificationAdapters:
     cycle_publisher: NotificationPublisher
     delivery_events: list[dict[str, Any]]
 
-    def publish_cycle_notification(self, *, detailed_text: str, compact_text: str) -> None:
-        self.cycle_publisher.publish(
+    def publish_cycle_notification(self, *, detailed_text: str, compact_text: str) -> bool:
+        before_count = len(self.delivery_events)
+        outcome = self.cycle_publisher.publish(
             RenderedNotification(
                 detailed_text=detailed_text,
                 compact_text=compact_text,
             )
         )
+        deliveries = self.delivery_events[before_count:]
+        if deliveries:
+            return all(event.get("delivery_status") == "sent" for event in deliveries)
+        return outcome is not False
 
 
 def build_runtime_notification_adapters(
@@ -51,17 +56,34 @@ def build_runtime_notification_adapters(
 ) -> LongBridgeNotificationAdapters:
     recorded_delivery_events = delivery_events if delivery_events is not None else []
 
-    def send_recorded_message(message: str) -> None:
-        send_message(message)
+    def send_recorded_message(message: str) -> bool:
         compact = str(message or "")
-        recorded_delivery_events.append(
+        event = {
+            "sink": notification_channel,
+            "compact_text_sha256": hashlib.sha256(compact.encode("utf-8")).hexdigest(),
+            "compact_text_length": len(compact),
+        }
+        try:
+            outcome = send_message(message)
+        except Exception as exc:
+            event.update(
+                {
+                    "delivery_status": "failed",
+                    "transport_acknowledged": False,
+                    "error_type": type(exc).__name__,
+                }
+            )
+            recorded_delivery_events.append(event)
+            return False
+        acknowledged = outcome is not False
+        event.update(
             {
-                "sink": notification_channel,
-                "delivery_status": "sent",
-                "compact_text_sha256": hashlib.sha256(compact.encode("utf-8")).hexdigest(),
-                "compact_text_length": len(compact),
+                "delivery_status": "sent" if acknowledged else "failed",
+                "transport_acknowledged": acknowledged,
             }
         )
+        recorded_delivery_events.append(event)
+        return acknowledged
 
     cycle_publisher = NotificationPublisher(
         log_message=log_message or (lambda message: print(with_prefix(message), flush=True)),

main.py

@@ -329,6 +329,32 @@ def _build_notification_delivery_log_for_report(
     }
 
 
+def _build_notification_delivery_summary(delivery_events: list[dict]) -> dict:
+    safe_fields = (
+        "sink",
+        "delivery_status",
+        "transport_acknowledged",
+        "error_type",
+        "compact_text_sha256",
+        "compact_text_length",
+    )
+    events = [
+        {key: event[key] for key in safe_fields if key in event}
+        for event in (dict(item) for item in delivery_events)
+    ]
+    if not events:
+        return {}
+    sent_count = sum(event.get("delivery_status") == "sent" for event in events)
+    failed_count = sum(event.get("delivery_status") == "failed" for event in events)
+    return {
+        "attempted_count": len(events),
+        "sent_count": sent_count,
+        "failed_count": failed_count,
+        "all_acknowledged": failed_count == 0 and sent_count == len(events),
+        "delivery_events": events,
+    }
+
+
 signal_text = build_signal_text(t)
 strategy_display_name = build_strategy_display_name(t)(
     STRATEGY_PROFILE,
@@ -516,16 +542,28 @@ def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
         print("LongBridge runtime error notification skipped: no Telegram target configured.", flush=True)
         return False
     message = _runtime_error_notification_message(exc, route_label=route_label)
+    outcomes = []
     for token, chat_id in targets:
         try:
-            requests.post(
+            response = requests.post(
                 f"https://api.telegram.org/bot{token}/sendMessage",
                 json={"chat_id": chat_id, "text": message},
                 timeout=10,
             )
+            status_code = int(getattr(response, "status_code", 200) or 200)
+            acknowledged = 200 <= status_code < 300
+            load_payload = getattr(response, "json", None)
+            payload = load_payload() if acknowledged and callable(load_payload) else None
+            if isinstance(payload, dict) and payload.get("ok") is False:
+                acknowledged = False
+            outcomes.append(acknowledged)
         except Exception as send_exc:
-            print(f"LongBridge runtime error Telegram send failed: {send_exc}", flush=True)
-    return True
+            print(
+                f"LongBridge runtime error Telegram send failed: {type(send_exc).__name__}",
+                flush=True,
+            )
+            outcomes.append(False)
+    return bool(outcomes) and all(outcomes)
 
 
 def _handle_route_runtime_error(exc: Exception, *, route_label: str):
@@ -585,7 +623,15 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
     composer = build_composer(dry_run_only_override=True if validation_only else None)
     reporting_adapters = composer.build_reporting_adapters()
     log_context, report = reporting_adapters.start_run()
-    notification_adapters = composer.build_notification_adapters()
+    notification_delivery_events: list[dict] = []
+    try:
+        notification_adapters = composer.build_notification_adapters(
+            delivery_events=notification_delivery_events,
+        )
+    except TypeError as exc:
+        if "delivery_events" not in str(exc):
+            raise
+        notification_adapters = composer.build_notification_adapters()
     strategy_plugin_signals, strategy_plugin_error = composer.load_strategy_plugin_signals(
         getattr(RUNTIME_SETTINGS, "strategy_plugin_mounts_json", None)
     )
@@ -679,7 +725,6 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             return True
         if not validation_only:
             publish_strategy_plugin_alerts(strategy_plugin_signals, report=report)
-        notification_delivery_events: list[dict] = []
         try:
             rebalance_runtime = composer.build_rebalance_runtime(
                 silent_cycle_notifications=validation_only,
@@ -722,6 +767,11 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
         )
         if notification_delivery_log:
             execution_summary["notification_delivery_log"] = notification_delivery_log
+        notification_delivery_summary = _build_notification_delivery_summary(
+            notification_delivery_events
+        )
+        if notification_delivery_summary:
+            execution_summary["notification_delivery_summary"] = notification_delivery_summary
         if signal_snapshot:
             reporting_adapters.log_event(
                 log_context,
@@ -749,7 +799,6 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             message=str(exc),
             error_type=type(exc).__name__,
         )
-        finalize_runtime_report(report, status="error")
         reporting_adapters.log_event(
             log_context,
             "strategy_cycle_failed",
@@ -759,9 +808,39 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             error_message=str(exc),
         )
         err = traceback.format_exc()
-        notification_adapters.publish_cycle_notification(
-            detailed_text=f"Strategy error:\n{err}",
-            compact_text=_compact_error_notification(exc),
+        try:
+            notification_adapters.publish_cycle_notification(
+                detailed_text=f"Strategy error:\n{err}",
+                compact_text=_compact_error_notification(exc),
+            )
+        except Exception as notification_exc:
+            notification_delivery_events.append(
+                {
+                    "sink": "telegram",
+                    "delivery_status": "failed",
+                    "transport_acknowledged": False,
+                    "error_type": type(notification_exc).__name__,
+                }
+            )
+            reporting_adapters.log_event(
+                log_context,
+                "strategy_error_notification_failed",
+                message="Strategy error notification failed",
+                severity="ERROR",
+                error_type=type(notification_exc).__name__,
+            )
+        error_summary = {}
+        notification_delivery_summary = _build_notification_delivery_summary(
+            notification_delivery_events
+        )
+        if notification_delivery_summary:
+            error_summary["notification_delivery_summary"] = (
+                notification_delivery_summary
+            )
+        finalize_runtime_report(
+            report,
+            status="error",
+            summary=error_summary or None,
         )
         return False
     finally:

notifications/telegram.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import re
+from collections.abc import Mapping
 from typing import Any
 
 from notifications.events import NotificationPublisher, RenderedNotification
@@ -489,19 +490,30 @@ def build_sender(token, chat_id, *, with_prefix_fn, requests_module=None):
     if requests_module is None:
         import requests as requests_module
 
-    def send_tg_message(message):
+    def send_tg_message(message) -> bool:
         if not token or not chat_id:
-            return
+            return False
         url = f"https://api.telegram.org/bot{token}/sendMessage"
         try:
             prefixed = with_prefix_fn(message)
-            requests_module.post(
+            response = requests_module.post(
                 url,
                 json={"chat_id": chat_id, "text": _break_telegram_market_symbol_auto_links(prefixed)},
                 timeout=10,
             )
+            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