fix: harden runtime scheduling and notifications

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

.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,6 +57,9 @@ 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_SCHEDULER_AWARE: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_AWARE || 'true' }}
       RUNTIME_HEARTBEAT_SCHEDULER_LOCATION: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_LOCATION || vars.CLOUD_RUN_REGION }}
       RUNTIME_HEARTBEAT_EXPECTED_DAY_OF_MONTH: ${{ vars.RUNTIME_HEARTBEAT_EXPECTED_DAY_OF_MONTH }}
@@ -74,6 +78,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 +99,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):
@@ -722,6 +760,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,

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
 

runtime_config_support.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import math
 import os
 from dataclasses import dataclass
@@ -235,6 +236,21 @@ def _market_defaults(market: str) -> dict[str, str]:
     }
 
 
+def _runtime_target_market_value(runtime_target: RuntimeTarget, field: str) -> str | None:
+    value = getattr(runtime_target, field, None)
+    if value is not None and str(value).strip():
+        return str(value).strip()
+    raw = os.getenv("QSL_RUNTIME_TARGET_JSON") or os.getenv("RUNTIME_TARGET_JSON")
+    if not raw:
+        return None
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        return None
+    value = payload.get(field) if isinstance(payload, dict) else None
+    return str(value).strip() if value is not None and str(value).strip() else None
+
+
 def load_platform_runtime_settings(
     *,
     project_id_resolver: Callable[[], str | None],
@@ -268,7 +284,13 @@ def load_platform_runtime_settings(
         os.getenv("ACCOUNT_REGION"),
         account_prefix=account_prefix,
     )
-    market = infer_market(os.getenv("LONGBRIDGE_MARKET"), account_region=account_region)
+    market = infer_market(
+        _first_non_empty(
+            os.getenv("LONGBRIDGE_MARKET"),
+            _runtime_target_market_value(runtime_target, "market"),
+        ),
+        account_region=account_region,
+    )
     market_defaults = _market_defaults(market)
     return PlatformRuntimeSettings(
         project_id=project_id_resolver(),
@@ -281,11 +303,13 @@ def load_platform_runtime_settings(
         market=market,
         market_calendar=_first_non_empty(
             os.getenv("LONGBRIDGE_MARKET_CALENDAR"),
+            _runtime_target_market_value(runtime_target, "market_calendar"),
             market_defaults["market_calendar"],
         )
         or DEFAULT_MARKET_CALENDAR,
         market_timezone=_first_non_empty(
             os.getenv("LONGBRIDGE_MARKET_TIMEZONE"),
+            _runtime_target_market_value(runtime_target, "market_timezone"),
             market_defaults["market_timezone"],
         )
         or DEFAULT_MARKET_TIMEZONE,