fix: centralize Telegram notification safety and redaction

README.md

@@ -140,3 +140,9 @@ See [CONTRIBUTING.md](CONTRIBUTING.md). AI assistants and maintainers follow [AG
 See [LICENSE](LICENSE).
 
 Display/notification naming configuration: [display names](docs/display-names.md).
+
+Telegram delivery uses QPK notification intents and redacted receipts. Runtime
+entrypoints install shared log and stream redaction. Environment-backed bot
+routes resolve credentials only in QPK delivery; existing injected or in-memory
+credentials use QPK's compatibility sender, with no temporary environment mutation.
+Provider errors and credential-bearing request data are excluded from reports.

main.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+
 from notifications.telegram import build_translator
 
 import hashlib
@@ -86,6 +87,9 @@
 from strategy_registry import get_platform_profile_status_matrix
 from strategy_runtime import load_strategy_runtime
 
+from quant_platform_kit.notifications import install_log_redaction
+install_log_redaction(wrap_streams=True)
+
 MARKET_CALENDAR = os.getenv("FIRSTRADE_MARKET_CALENDAR", "NYSE")
 MARKET_TIMEZONE = os.getenv("FIRSTRADE_MARKET_TIMEZONE", "America/New_York")
 

notifications/intent_transport.py

@@ -0,0 +1,87 @@
+"""Adapt existing injected HTTP clients to the shared QPK Telegram sender."""
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import os
+
+from quant_platform_kit.notifications import (
+    NotificationIntent,
+    redact_sensitive_text,
+    send_notification_intent,
+    send_telegram_message,
+)
+
+
+def send_intent(text, *, chat_id, token, token_env="TELEGRAM_TOKEN", requests_module,
+                timeout=15, payload_key="json", request_options=None, printer=None):
+    """Keep credentials outside intents, reports and provider error receipts.
+
+    Environment-backed runtime routes use the intent sender. Existing callers
+    supplying an in-memory credential retain QPK's explicit-credential transport
+    until they can provide a credential reference; no process environment is mutated.
+    """
+    message = redact_sensitive_text(str(text))
+    intent = NotificationIntent(text=message, run_id="runtime", chat_ids=(str(chat_id),))
+    provider = {}
+    statuses = []
+
+    def opener(request, *, timeout):
+        payload = json.loads(request.data)
+        # Preserve existing consumer wire options; QPK still builds/validates the request.
+        payload.pop("disable_web_page_preview", None)
+        try:
+            response = requests_module.post(
+                request.full_url, **{payload_key: payload}, timeout=timeout,
+                **(request_options or {}),
+            )
+        except Exception as exc:
+            error_type = type(exc).__name__
+            provider["error_type"] = error_type if error_type.isidentifier() else "transport_error"
+            raise
+        status = int(getattr(response, "status_code", 200) or 200)
+        statuses.append(status)
+        if printer is not None and not 200 <= status < 300:
+            detail = redact_sensitive_text(str(getattr(response, "text", "")))
+            if token:
+                detail = detail.replace(str(token), "<redacted>")
+            printer(f"Telegram send failed with status {status}: {detail[:300]}", flush=True)
+        load = getattr(response, "json", None)
+        body = load() if callable(load) else None
+        if isinstance(body, dict):
+            # Retain only scalar delivery metadata needed by existing consumers.
+            result = body.get("result")
+            if body.get("ok") is True and isinstance(result, dict):
+                message_id = result.get("message_id")
+                if type(message_id) is int and message_id > 0:
+                    provider["message_id"] = message_id
+                    chat = result.get("chat")
+                    if isinstance(chat, dict) and str(chat.get("id")) == str(chat_id):
+                        provider["transport_receipt_id"] = hashlib.sha256(
+                            json.dumps([str(chat_id), message_id], separators=(",", ":")).encode()
+                        ).hexdigest()
+            parameters = body.get("parameters")
+            delay = parameters.get("retry_after") if isinstance(parameters, dict) else None
+            if type(delay) is int and delay > 0:
+                provider["retry_after"] = delay
+        wrapped = io.BytesIO(json.dumps(body).encode())
+        wrapped.status = status
+        return wrapped
+
+    if token == os.environ.get(token_env):
+        receipt = send_notification_intent(intent, bot_token_env=token_env, opener=opener, timeout=timeout)
+    else:
+        # Compatibility for existing dependency-injected and SecretStore callers.
+        # QPK remains the sole URL/HTTP implementation, with the same strict ack.
+        ok = send_telegram_message(
+            bot_token=token, chat_ids=intent.chat_ids, text=intent.text,
+            opener=opener, timeout=timeout, printer=lambda *_args, **_kwargs: None,
+        )
+        receipt = {
+            "ok": ok, "status": "sent" if ok else "failed",
+            "http_status": statuses[-1] if statuses else None,
+            "sha256": hashlib.sha256(message.encode()).hexdigest(),
+            "length": len(message), "run_id": intent.run_id,
+        }
+    return {**receipt, **provider}

notifications/telegram.py

@@ -527,26 +527,11 @@ def build_sender(token: str | None, chat_id: str | None, *, requests_module=None
     def send_tg_message(message: str) -> bool:
         if not token or not chat_id:
             return False
-        url = f"https://api.telegram.org/bot{token}/sendMessage"
-        try:
-            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, 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
+        from .intent_transport import send_intent
+        return send_intent(
+            message, chat_id=chat_id, token=token,
+            requests_module=requests_module, timeout=15,
+        )["ok"]
 
     return send_tg_message
 

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "pytest",
     "pytz",
     "requests",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@5d0714201f3813f68a9833083cd21f6d61e5b702",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@c31f28484b8489a1ecda8e595a6d2afd3bce8185",
 ]
 license = "MIT"
@@ -82,5 +82,5 @@ show_missing = true
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@5d0714201f3813f68a9833083cd21f6d61e5b702",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5",
 ]

qsl.toml

@@ -5,7 +5,7 @@ upgrade_ring = "ring_d"
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "5d0714201f3813f68a9833083cd21f6d61e5b702"
+quant_platform_kit = "59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5"
 us_equity_strategies = "c31f28484b8489a1ecda8e595a6d2afd3bce8185"
 
 [qsl.compat]

scripts/notify_route_check.py

@@ -123,6 +123,8 @@ def _accessor_members(secret: str, project: str) -> dict:
 
 
 def main() -> int:
+    from quant_platform_kit.notifications import install_log_redaction
+    install_log_redaction()
     project = os.environ["GCP_PROJECT_ID"]
     prefix = (os.environ.get("SERVICE_NAME_PREFIX") or "").strip()
     limit = int(os.environ.get("REVISION_LIMIT") or "5")

tests/test_notification_off_equivalence.py

@@ -172,6 +172,17 @@ def test_flags_off_entrypoints_renderers_match_main_golden(monkeypatch, flag, mo
         for outcome in ("blocked", "missing"):
             for notify in (False, True):
                 key = f"heartbeat:{lang}:{outcome}:{notify}"
+                # The pinned QPK privacy contract replaces raw technical details.
+                # Keep the historical golden immutable and compare every control call.
+                if outcome == "blocked":
+                    raw = ("技术详情（原文）：\nsynthetic-service: rejected inner execution status"
+                           if lang == "zh" else
+                           "Technical details (original):\nsynthetic-service: rejected inner execution status")
+                    safe = ("排查详情请查看工作流或云端运行日志。" if lang == "zh" else
+                            "See the workflow or cloud runtime logs for diagnostic details.")
+                    send_call = expected[key]["calls"][-1]
+                    assert send_call[0] == "send" and send_call[1].endswith(raw)
+                    send_call[1] = send_call[1][:-len(raw)] + safe
                 assert actual[key] == expected[key]
     actual_broker = actual["broker_cycle"]
     expected_broker = expected["broker_cycle"]

tests/test_notification_quiet.py

@@ -152,7 +152,8 @@ def read(uri, **kw):
     monkeypatch.setattr(heartbeat, '_cat_gcs_json', read)
     assert heartbeat.main(NOW) == 0
     assert len(sent) == 1
-    assert 'unknown' in sent[0]
+    assert 'Runtime check evidence needs review' in sent[0]
+    assert 'See the workflow or cloud runtime logs' in sent[0]
 
 
 def test_conflicting_latest_identity_cannot_be_hidden_by_old_healthy(monkeypatch):
@@ -170,7 +171,9 @@ def read(uri, **kw):
     monkeypatch.setattr(heartbeat, '_cat_gcs_json', read)
     assert heartbeat.main(NOW) == 1
     assert len(sent) == 1
-    assert 'conflicting report identity' in sent[0]
+    assert 'None of the most recent 2 execution reports was acceptable' in sent[0]
+    assert 'conflicting report identity' not in sent[0]
+    assert 'See the workflow or cloud runtime logs' in sent[0]
 
 
 @pytest.mark.parametrize('value', [True, 1, 'bad'])
@@ -337,4 +340,5 @@ def test_reuploaded_old_payload_is_visible_only_against_existing_precise_due(mon
     assert bool(sent) is stale
     if stale:
         assert heartbeat._notice('status_normal') not in sent[0]
-        assert 'predates latest due' in sent[0]
+        assert 'Runtime check evidence needs review' in sent[0]
+        assert 'See the workflow or cloud runtime logs' in sent[0]

tests/test_report_serialization_no_secrets.py

@@ -0,0 +1,52 @@
+import json
+import os
+import re
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from notifications.intent_transport import send_intent
+from quant_platform_kit.common.runtime_reports import build_runtime_report_base
+
+
+def test_report_serialization_has_no_secrets():
+    token = str(123456789) + ':' + 'A' * 35
+    chat_id = 'synthetic-private-chat'
+    for acknowledged in (True, False):
+        transport = Mock()
+        transport.post.return_value = SimpleNamespace(status_code=200, json=lambda: {'ok': acknowledged})
+        with patch.dict(os.environ, {'TELEGRAM_TOKEN': token}, clear=True):
+            receipt = send_intent('Synthetic cycle status', chat_id=chat_id, token=token,
+                                  requests_module=transport)
+        report = build_runtime_report_base(
+            platform='FirstradePlatform', deploy_target='synthetic', service_name='synthetic',
+            strategy_profile='synthetic', run_id='synthetic', run_source='test',
+            summary={'notification_receipt': receipt},
+        )
+        serialized = json.dumps(report)
+        assert receipt['ok'] is acknowledged
+        assert transport.post.call_count == 1
+        assert token not in serialized
+        assert chat_id not in serialized
+        assert re.search(r'\d{6,12}:A[\w-]{34}', serialized) is None
+
+        def check(value):
+            if isinstance(value, dict):
+                for key, item in value.items():
+                    if key.lower() in {'token', 'secret', 'password', 'api_key'}:
+                        assert not item
+                    check(item)
+            elif isinstance(value, list):
+                for item in value:
+                    check(item)
+        check(json.loads(serialized))
+
+
+def test_provider_exception_cannot_enter_report():
+    token = str(123456789) + ':' + 'A' * 35
+    transport = Mock()
+    transport.post.side_effect = RuntimeError(token)
+    with patch.dict(os.environ, {'TELEGRAM_TOKEN': token}, clear=True):
+        receipt = send_intent('Synthetic cycle status', chat_id='synthetic-private-chat',
+                              token=token, requests_module=transport)
+    assert not receipt['ok']
+    assert token not in json.dumps(receipt)