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: charlesschwabquant
@@ -44,16 +45,34 @@ 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.SCHWAB_MARKET_CALENDAR }}
+      RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.SCHWAB_MARKET_TIMEZONE }}
       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 }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
     steps:
       - name: Checkout repository
         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 }}
@@ -62,5 +81,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

@@ -64,6 +64,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/rebalance_service.py

@@ -52,6 +52,32 @@ def _record_platform_execution_telemetry(
     )
 
 
+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 _plan_portfolio(plan):
     return dict(plan.get("portfolio") or {})
 
@@ -596,5 +622,10 @@ def load_plan(current_snapshot):
             + json.dumps({"reason": "no_trade_or_error", "trade_logs_count": len(trade_logs)}, ensure_ascii=False),
             flush=True,
         )
+    notification_delivery_summary = _build_notification_delivery_summary(
+        getattr(runtime, "notification_delivery_events", []) or []
+    )
+    if notification_delivery_summary:
+        execution["notification_delivery_summary"] = notification_delivery_summary
     _record_platform_execution_telemetry(config, execution_result)
     return execution_result

application/runtime_composer.py

@@ -57,26 +57,25 @@ class SchwabRuntimeComposer:
     printer: Callable[..., Any] = print
     notification_channel: str = "telegram"
     webhook_url: str | None = None
-    sender_builder: Callable[..., Callable[[str], None]] | None = None
+    sender_builder: Callable[..., Callable[[str], bool | None]] | None = None
     notification_builder: Callable[..., Any] = build_runtime_notification_adapters
     reporting_builder: Callable[..., Any] = build_runtime_reporting_adapters
     runtime_target: RuntimeTarget | None = None
     limit_buy_premium_by_symbol: dict[str, float] | None = None
     extra_reporting_fields: dict[str, Any] = field(default_factory=dict)
 
-    def send_message(self, message: str) -> None:
+    def send_message(self, message: str) -> bool:
         """Send a cycle notification through the configured channel."""
         if self.sender_builder is not None:
             sender = self.sender_builder(self.tg_token, self.tg_chat_id)
-            sender(message)
-            return
+            return sender(message) is not False
         sender = build_cycle_sender(
             channel=self.notification_channel,
             telegram_token=self.tg_token,
             telegram_chat_id=self.tg_chat_id,
             webhook_url=self.webhook_url,
         )
-        sender(message)
+        return sender(message)
 
     send_tg_message = send_message  # backward-compat alias
 
@@ -152,6 +151,11 @@ def build_rebalance_runtime(self, client, *, silent_cycle_notifications: bool =
                 client,
                 account_hash,
             ),
+            notification_delivery_events=getattr(
+                notification_adapters,
+                "delivery_events",
+                [],
+            ),
         )
 
     def build_rebalance_config(

application/runtime_dependencies.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any
 
 from quant_platform_kit.common.ports import ExecutionPort, MarketDataPort, NotificationPort, PortfolioPort
@@ -42,3 +42,4 @@ class SchwabRebalanceRuntime:
     execution_port_factory: Callable[[str], ExecutionPort] | None = None
     order_status_fetcher_factory: Callable[[str], Callable[[str], Any] | None] | None = None
     submit_equity_order: Callable[..., Any] | None = None
+    notification_delivery_events: list[dict[str, Any]] = field(default_factory=list)

application/runtime_notification_adapters.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import hashlib
 from dataclasses import dataclass
+from typing import Any
 
 from notifications.events import NotificationPublisher, RenderedNotification
 from quant_platform_kit.common.port_adapters import CallableNotificationPort
@@ -13,26 +15,65 @@
 class SchwabNotificationAdapters:
     notification_port: NotificationPort
     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(
     *,
     send_message,
     notification_channel: str = "telegram",
     log_message=None,
+    delivery_events: list[dict[str, Any]] | None = None,
 ) -> SchwabNotificationAdapters:
+    recorded_delivery_events = delivery_events if delivery_events is not None else []
+
+    def send_recorded_message(message: str) -> bool:
+        compact = str(message or "")
+        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(
+            {
+                "delivery_status": "sent" if acknowledged else "failed",
+                "transport_acknowledged": acknowledged,
+            }
+        )
+        recorded_delivery_events.append(event)
+        return acknowledged
+
     return SchwabNotificationAdapters(
-        notification_port=CallableNotificationPort(send_message),
+        notification_port=CallableNotificationPort(send_recorded_message),
         cycle_publisher=NotificationPublisher(
             log_message=log_message or (lambda message: print(message, flush=True)),
-            send_message=send_message,
+            send_message=send_recorded_message,
         ),
+        delivery_events=recorded_delivery_events,
     )

application/runtime_report_summary.py

@@ -52,4 +52,7 @@ def summarize_execution_cycle_result(result: object, *, dry_run: bool) -> dict[s
         value = execution.get(field_name)
         if value is not None and value != "":
             summary[field_name] = value
+    notification_delivery_summary = _as_mapping(execution.get("notification_delivery_summary"))
+    if notification_delivery_summary:
+        summary["notification_delivery_summary"] = notification_delivery_summary
     return summary

main.py

@@ -407,7 +407,7 @@ def send_tg_message(message):
 
 
 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,
     )
@@ -469,22 +469,35 @@ def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
         print("Schwab 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=15,
             )
+            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"Schwab runtime error Telegram send failed: {send_exc}", flush=True)
-    return True
+            print(
+                f"Schwab runtime error Telegram send failed: {type(send_exc).__name__}",
+                flush=True,
+            )
+            outcomes.append(False)
+    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, route_label="strategy_cycle")
     except Exception as notification_exc:
         print(f"Schwab runtime error notification fallback: {notification_exc}", flush=True)
         return _notify_runtime_error(exc, route_label="strategy_cycle")

notifications/telegram.py

@@ -417,17 +417,28 @@ def build_sender(token, chat_id, *, 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:
-            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, dict) 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