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: interactivebrokersquant
@@ -44,13 +45,18 @@ 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.IBKR_MARKET_CALENDAR }}
+      RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.IBKR_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_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 }}
@@ -59,6 +65,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 }}
@@ -67,5 +86,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

@@ -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 }}

application/runtime_notification_adapters.py

@@ -17,13 +17,18 @@ class IBKRNotificationAdapters:
     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(
@@ -35,17 +40,34 @@ def build_runtime_notification_adapters(
 ) -> IBKRNotificationAdapters:
     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
 
     return IBKRNotificationAdapters(
         notification_port=CallableNotificationPort(send_recorded_message),

entrypoints/cloud_run.py

@@ -15,7 +15,7 @@ def _load_market_calendar(calendar_name: str, *, logger) -> object | None:
     except Exception as exc:
         logger(
             f"pandas_market_calendars unavailable for {calendar_name}: {exc}; "
-            "falling back to weekday-only market-open check"
+            "failing closed as market closed"
         )
         return None
 
@@ -31,7 +31,7 @@ def is_market_open_now(
     now_ny = datetime.now(tz_ny)
     calendar = _load_market_calendar(calendar_name, logger=logger)
     if calendar is None:
-        return now_ny.weekday() < 5
+        return False
     schedule = calendar.schedule(start_date=now_ny.date(), end_date=now_ny.date())
     if len(getattr(schedule, "index", ())) == 0:
         return False
@@ -54,6 +54,6 @@ def is_market_open_today(
     now_ny = datetime.now(tz_ny)
     calendar = _load_market_calendar(calendar_name, logger=logger)
     if calendar is None:
-        return now_ny.weekday() < 5
+        return False
     schedule = calendar.schedule(start_date=now_ny.date(), end_date=now_ny.date())
     return len(getattr(schedule, "index", ())) > 0

main.py

@@ -701,7 +701,7 @@ def send_tg_message(message):
         telegram_chat_id=TG_CHAT_ID,
         webhook_url=_NOTIFICATION_WEBHOOK_URL,
     )
-    sender(_with_platform_notification_prefix(message))
+    return sender(_with_platform_notification_prefix(message))
 
 
 def _platform_notification_prefix() -> str:
@@ -739,7 +739,7 @@ def _with_platform_notification_prefix(message: str) -> str:
 
 
 def publish_notification(*, detailed_text, compact_text):
-    build_composer().build_notification_adapters().publish_cycle_notification(
+    return build_composer().build_notification_adapters().publish_cycle_notification(
         detailed_text=detailed_text,
         compact_text=compact_text,
     )
@@ -796,20 +796,22 @@ def _notify_runtime_error(exc: Exception, *, route_label: str | None = None) ->
         print("IBKR 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:
-        send_telegram_message(
+        outcomes.append(send_telegram_message(
             _with_platform_notification_prefix(message),
             token=token,
             chat_id=chat_id,
             requests_module=requests,
-        )
-    return True
+        ))
+    return bool(outcomes) and all(outcomes)
 
 
 def _publish_runtime_failure_notification(*, detailed_text: str, compact_text: str, exc: Exception) -> bool:
     try:
-        publish_notification(detailed_text=detailed_text, compact_text=compact_text)
-        return True
+        if publish_notification(detailed_text=detailed_text, compact_text=compact_text):
+            return True
+        return _notify_runtime_error(exc)
     except Exception as notification_exc:
         print(f"IBKR runtime error notification fallback: {notification_exc}", flush=True)
         return _notify_runtime_error(exc)
@@ -1025,6 +1027,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,
+    }
+
+
 def publish_strategy_plugin_alerts(signals, *, report=None):
     result = dispatch_strategy_plugin_alerts(
         signals,
@@ -1329,6 +1357,11 @@ def _handle_request(
         )
         if notification_delivery_log:
             report_summary["notification_delivery_log"] = notification_delivery_log
+        notification_delivery_summary = _build_notification_delivery_summary(
+            notification_delivery_events
+        )
+        if notification_delivery_summary:
+            report_summary["notification_delivery_summary"] = notification_delivery_summary
         execution_status = str(report_summary.get("execution_status") or "").strip().lower()
         execution_failed = _is_execution_failure(
             execution_status,

notifications/telegram.py

@@ -523,7 +523,7 @@ def send_telegram_message(
     printer=print,
 ):
     if not token or not chat_id:
-        return
+        return False
 
     url = f"https://api.telegram.org/bot{token}/sendMessage"
     try:
@@ -539,5 +539,13 @@ def send_telegram_message(
                 f"{response.status_code}: {_safe_telegram_error_text(response.text, token=token)}",
                 flush=True,
             )
+            return False
+        load_payload = getattr(response, "json", None)
+        payload = load_payload() if callable(load_payload) else None
+        if isinstance(payload, dict) and payload.get("ok") is False:
+            printer("Telegram send failed: negative API acknowledgement", flush=True)
+            return False
     except Exception as exc:
         printer(f"Telegram send failed: {type(exc).__name__}", flush=True)
+        return False
+    return True

runtime_config_support.py

@@ -257,6 +257,21 @@ class PlatformRuntimeSettings:
     execution_backend: str = EXECUTION_BACKEND_GATEWAY
 
 
+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],
@@ -336,7 +351,13 @@ def load_platform_runtime_settings(
         )
         ib_client_id = group_config.ib_client_id or 0
 
-    market = resolve_market(os.getenv("IBKR_MARKET"), account_group=account_group)
+    market = resolve_market(
+        first_non_empty(
+            os.getenv("IBKR_MARKET"),
+            _runtime_target_market_value(runtime_target, "market"),
+        ),
+        account_group=account_group,
+    )
     market_defaults = market_default_settings(market)
     return PlatformRuntimeSettings(
         project_id=project_id,
@@ -386,6 +407,7 @@ def load_platform_runtime_settings(
         market=market,
         market_calendar=first_non_empty(
             os.getenv("IBKR_MARKET_CALENDAR"),
+            _runtime_target_market_value(runtime_target, "market_calendar"),
             market_defaults["market_calendar"],
         ),
         market_currency=first_non_empty(
@@ -404,6 +426,7 @@ def load_platform_runtime_settings(
         ).upper(),
         market_timezone=first_non_empty(
             os.getenv("IBKR_MARKET_TIMEZONE"),
+            _runtime_target_market_value(runtime_target, "market_timezone"),
             market_defaults["market_timezone"],
         ),
         quantity_step=1.0,