Use platform Telegram token secrets for alerts

Author: Pigbibi

.github/workflows/execution-report-heartbeat.yml

@@ -46,6 +46,7 @@ jobs:
       FIRSTRADE_STATE_PREFIX: ${{ vars.FIRSTRADE_STATE_PREFIX }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
+      TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6

.github/workflows/runtime-guard.yml

@@ -54,6 +54,7 @@ jobs:
       CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
+      TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6

scripts/cloud_run_runtime_guard.py

@@ -85,6 +85,10 @@ def _load_services() -> list[str]:
     return unique
 
 
+def _run_gcloud(args: list[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(args, text=True, capture_output=True, check=False)
+
+
 def _run_gcloud_logging(project: str, log_filter: str, limit: int) -> list[dict[str, Any]]:
     command = [
         "gcloud",
@@ -166,10 +170,46 @@ def _summarize(entry: dict[str, Any]) -> str:
     return f"- {timestamp} {target or '<unknown>'} severity={severity}{status_text}{suffix}"
 
 
+
+
+def _telegram_secret_project() -> str | None:
+    return (
+        os.environ.get("RUNTIME_HEARTBEAT_GCP_PROJECT_ID")
+        or os.environ.get("RUNTIME_GUARD_GCP_PROJECT_ID")
+        or os.environ.get("GCP_PROJECT_ID")
+        or os.environ.get("GOOGLE_CLOUD_PROJECT")
+    )
+
+
+def _load_telegram_token_from_secret() -> str:
+    secret_name = (os.environ.get("TELEGRAM_TOKEN_SECRET_NAME") or "").strip()
+    if not secret_name:
+        return ""
+    command = ["gcloud", "secrets", "versions", "access", "latest", "--secret", secret_name]
+    project = _telegram_secret_project()
+    if project:
+        command.extend(["--project", project])
+    result = _run_gcloud(command)
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        print(
+            f"Unable to read Telegram token from Secret Manager: {detail or 'gcloud failed'}",
+            file=sys.stderr,
+        )
+        return ""
+    return result.stdout.strip()
+
+
+def _telegram_token() -> str:
+    direct_token = (os.environ.get("TELEGRAM_TOKEN") or os.environ.get("TG_TOKEN") or "").strip()
+    if direct_token:
+        return direct_token
+    return _load_telegram_token_from_secret()
+
 def _send_telegram(message: str) -> bool:
     targets: list[tuple[str, str]] = []
 
-    token = os.environ.get("TELEGRAM_TOKEN") or os.environ.get("TG_TOKEN")
+    token = _telegram_token()
     for chat_id in _split_values(os.environ.get("GLOBAL_TELEGRAM_CHAT_ID")):
         if token:
             targets.append((token, chat_id))

scripts/execution_report_heartbeat.py

@@ -289,9 +289,45 @@ def _is_accepted_report(payload: dict[str, Any]) -> tuple[bool, str]:
     return False, f"unaccepted status={status or '-'} stage={stage or '-'}"
 
 
+
+
+def _telegram_secret_project() -> str | None:
+    return (
+        os.environ.get("RUNTIME_HEARTBEAT_GCP_PROJECT_ID")
+        or os.environ.get("RUNTIME_GUARD_GCP_PROJECT_ID")
+        or os.environ.get("GCP_PROJECT_ID")
+        or os.environ.get("GOOGLE_CLOUD_PROJECT")
+    )
+
+
+def _load_telegram_token_from_secret() -> str:
+    secret_name = (os.environ.get("TELEGRAM_TOKEN_SECRET_NAME") or "").strip()
+    if not secret_name:
+        return ""
+    command = ["gcloud", "secrets", "versions", "access", "latest", "--secret", secret_name]
+    project = _telegram_secret_project()
+    if project:
+        command.extend(["--project", project])
+    result = _run_gcloud(command)
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        print(
+            f"Unable to read Telegram token from Secret Manager: {detail or 'gcloud failed'}",
+            file=sys.stderr,
+        )
+        return ""
+    return result.stdout.strip()
+
+
+def _telegram_token() -> str:
+    direct_token = (os.environ.get("TELEGRAM_TOKEN") or os.environ.get("TG_TOKEN") or "").strip()
+    if direct_token:
+        return direct_token
+    return _load_telegram_token_from_secret()
+
 def _send_telegram(message: str) -> bool:
     targets: list[tuple[str, str]] = []
-    token = os.environ.get("TELEGRAM_TOKEN") or os.environ.get("TG_TOKEN")
+    token = _telegram_token()
     for chat_id in _split_values(os.environ.get("GLOBAL_TELEGRAM_CHAT_ID")):
         if token:
             targets.append((token, chat_id))

tests/test_cloud_run_runtime_guard.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import subprocess
+
+from scripts import cloud_run_runtime_guard as guard
+
+
+def test_telegram_token_falls_back_to_secret_manager(monkeypatch):
+    monkeypatch.delenv("TELEGRAM_TOKEN", raising=False)
+    monkeypatch.delenv("TG_TOKEN", raising=False)
+    monkeypatch.setenv("TELEGRAM_TOKEN_SECRET_NAME", "platform-telegram-token")
+    monkeypatch.setenv("GCP_PROJECT_ID", "firstradequant")
+    observed = {}
+
+    def fake_run_gcloud(command):
+        observed["command"] = command
+        return subprocess.CompletedProcess(command, 0, stdout="secret-token\n", stderr="")
+
+    monkeypatch.setattr(guard, "_run_gcloud", fake_run_gcloud)
+
+    assert guard._telegram_token() == "secret-token"
+    assert observed["command"] == [
+        "gcloud",
+        "secrets",
+        "versions",
+        "access",
+        "latest",
+        "--secret",
+        "platform-telegram-token",
+        "--project",
+        "firstradequant",
+    ]

tests/test_execution_report_heartbeat.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import subprocess
 import datetime as dt
 
 from scripts import execution_report_heartbeat as heartbeat
@@ -24,3 +25,30 @@ def test_report_globs_include_sanitized_month_segments(monkeypatch):
         "gs://runtime-state/firstrade-platform/strategy-runs/**/2026-06/*.json",
         "gs://runtime-state/firstrade-platform/strategy-runs/**/2026_06/*.json",
     ]
+
+def test_telegram_token_falls_back_to_secret_manager(monkeypatch):
+    monkeypatch.delenv("TELEGRAM_TOKEN", raising=False)
+    monkeypatch.delenv("TG_TOKEN", raising=False)
+    monkeypatch.setenv("TELEGRAM_TOKEN_SECRET_NAME", "platform-telegram-token")
+    monkeypatch.setenv("GCP_PROJECT_ID", "firstradequant")
+    observed = {}
+
+    def fake_run_gcloud(command):
+        observed["command"] = command
+        return subprocess.CompletedProcess(command, 0, stdout="secret-token\n", stderr="")
+
+    monkeypatch.setattr(heartbeat, "_run_gcloud", fake_run_gcloud)
+
+    assert heartbeat._telegram_token() == "secret-token"
+    assert observed["command"] == [
+        "gcloud",
+        "secrets",
+        "versions",
+        "access",
+        "latest",
+        "--secret",
+        "platform-telegram-token",
+        "--project",
+        "firstradequant",
+    ]
+