fix: centralize Telegram sends with credential-free intents

README.md

@@ -131,3 +131,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.

logs.json

@@ -0,0 +1 @@
+[{"jsonPayload": {"event": "other", "status": "applied_paused", "readback": {"application_id": "61743a15-6dcc-4ebf-8ecb-454660014af7", "revision_name": "paper-v7-00001", "runtime_target_enabled": false}}}]

main.py

@@ -87,6 +87,9 @@
 from strategy_runtime import load_strategy_runtime
 from decision_mapper import map_strategy_decision_to_plan
 
+from quant_platform_kit.notifications import install_log_redaction
+install_log_redaction(wrap_streams=True)
+
 app = Flask(__name__)
 
 # This collector is limited to documented LongPort read endpoints and is
@@ -676,18 +679,10 @@ def _send_runtime_error(exc: Exception, *, route_label: str) -> bool:
     outcomes = []
     for token, chat_id in targets:
         try:
-            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)
+            from notifications.intent_transport import send_intent
+            outcomes.append(send_intent(
+                message, token=token, chat_id=chat_id, requests_module=requests, timeout=10,
+            )["ok"])
         except Exception:
             print(
                 t("runtime_notification_delivery_failed"),

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

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

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "google-cloud-storage",
     "google-auth",
     "longport==3.0.23",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@350dd38ece0952beef096893a7dcf8ae3871609a",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@8412d8338879f335ef8830678c2733e02e53e7c7",
     "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@fe0c32e09d454ca4388c3372bf3874097b1bbaf0",
 ]
@@ -61,5 +61,5 @@ include = [
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@350dd38ece0952beef096893a7dcf8ae3871609a",
+  "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 = "350dd38ece0952beef096893a7dcf8ae3871609a"
+quant_platform_kit = "59469a16b4ff7c5a1055caab21a1bd7a21dc0ea5"
 us_equity_strategies = "8412d8338879f335ef8830678c2733e02e53e7c7"
 hk_equity_strategies = "cc3045689ae9d76eb97b9e1aa3b8fb6288ee1bcc"
 

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")

scripts/paper_observation/__main__.py

@@ -150,20 +150,19 @@ def sender(token, recipient):
 
     def send(message):
         try:
-            response = requests.post(
-                f"https://api.telegram.org/bot{token}/sendMessage",
-                json={"chat_id": recipient, "text": message},
-                timeout=10,
-                allow_redirects=False,
-            )
-            return response.status_code == 200 and response.json().get("ok") is True
+            from notifications.intent_transport import send_intent
+            receipt = send_intent(message, token=token, chat_id=recipient, requests_module=requests,
+                                  timeout=10, request_options={"allow_redirects": False})
+            return receipt["http_status"] == 200 and receipt["ok"]
         except Exception:
             return False
 
     return send
 
 
 def main():
+    from quant_platform_kit.notifications import install_log_redaction
+    install_log_redaction(wrap_streams=True)
     logging.disable(logging.CRITICAL)
     try:
         if len(sys.argv) != 2 or sys.argv[1] not in ("collect", "watchdog", "verify"):

tests/test_durable_execution_commands.py

@@ -163,7 +163,8 @@ def enqueue(self, _command):
     assert gate["enforcement"] == "observe"
     assert gate["mode"] == "active"
     assert gate["policy_allows"] is False
-    assert gate["broker_write_allowed"] is True
+    # The pinned QPK command gate never treats observation as broker authority.
+    assert gate["broker_write_allowed"] is False
     assert "exposure_effect_unknown" in gate["reasons"]
     assert result["consumer_authorized"] is False
 

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='LongBridgePlatform', 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)

tests/test_runtime_strategy_adapters.py

@@ -512,7 +512,7 @@ def test_runtime_strategy_adapters_builds_escalated_plugin_alert_message():
     alerts = adapters.build_strategy_plugin_alert_messages((signal,))
 
     assert len(alerts) == 1
-    assert alerts[0].subject == "alert:soxl_soxx_trend_income:Crisis:true crisis"
+    assert alerts[0].subject == "alert:Semiconductor Trend Income:Crisis:true crisis"
     assert "plugin=Crisis" in alerts[0].body
     assert "route=true crisis" in alerts[0].body
     assert "action=defend" in alerts[0].body