Split plugin alerts from strategy position notices

Author: Pigbibi

docs/strategy_plugin_runtime_contract.md

@@ -177,11 +177,22 @@ when any of the following is true:
 - `suggested_action` is `defend` or `blocked`
 - `would_trade_if_enabled` is `true`
 
+Strategy-mounted artifacts that are automation-approved, expose
+`position_control_allowed = true`, and request an automatic `defend` or
+`delever` action are intentionally excluded from the dedicated plugin-alert
+stream. Those position-impacting events should be reported by the strategy run
+that consumed the artifact. The plugin-alert stream remains for manual-review
+or notification-only cases, including `notification_targets`, `blocked`,
+`watch_only`, and `notify_manual_review` routes.
+
 Platforms may still choose their delivery sinks, but shared escalation helpers
-are available for email and SMS:
+are available for email, SMS, push, and Telegram:
 
+- `quant_platform_kit.notifications.strategy_plugin_alerts.publish_strategy_plugin_alerts()`
 - `quant_platform_kit.notifications.strategy_plugin_email.publish_strategy_plugin_email_alerts()`
 - `quant_platform_kit.notifications.strategy_plugin_sms.publish_strategy_plugin_sms_alerts()`
+- `quant_platform_kit.notifications.strategy_plugin_push.publish_strategy_plugin_push_alerts()`
+- `quant_platform_kit.notifications.strategy_plugin_telegram.publish_strategy_plugin_telegram_alerts()`
 
 The publishers build the shared subject/body, prefix platform context, return
 structured sent/skipped/failed diagnostics, and can use marker stores to skip

docs/strategy_plugin_runtime_contract.zh-CN.md

@@ -162,12 +162,21 @@ sidecar 路径维护插件账本或执行插件驱动的 allocation 变更。
 - `suggested_action` 是 `defend` 或 `blocked`
 - `would_trade_if_enabled` 是 `true`
 
+如果 strategy-mounted artifact 已经是 `automation_approved`、暴露
+`position_control_allowed = true`，并且请求自动 `defend` 或 `delever`
+动作，则专用插件告警流会刻意跳过它。这类会影响仓位的事件应由实际消费该
+artifact 的策略运行结果通知。插件告警流只保留给人工复核或 notification-only
+场景，包括 `notification_targets`、`blocked`、`watch_only` 和
+`notify_manual_review` 路线。
+
 平台仍可选择自己的投递 sink；共享 helper 已提供 email、SMS、push 和
 Telegram 的聚合入口：
 
 - `quant_platform_kit.notifications.strategy_plugin_alerts.publish_strategy_plugin_alerts()`
 - `quant_platform_kit.notifications.strategy_plugin_email.publish_strategy_plugin_email_alerts()`
 - `quant_platform_kit.notifications.strategy_plugin_sms.publish_strategy_plugin_sms_alerts()`
+- `quant_platform_kit.notifications.strategy_plugin_push.publish_strategy_plugin_push_alerts()`
+- `quant_platform_kit.notifications.strategy_plugin_telegram.publish_strategy_plugin_telegram_alerts()`
 
 publisher 会构造共享 subject/body、追加平台上下文、返回结构化
 sent/skipped/failed diagnostics，并可使用 marker store 跳过某个通道已发送过的

src/quant_platform_kit/common/strategy_plugins.py

@@ -32,6 +32,7 @@
 DEFAULT_PLUGIN_ARTIFACT_CACHE_DIR = Path(tempfile.gettempdir()) / "quant_strategy_plugin_artifacts"
 STRATEGY_PLUGIN_NON_ALERT_ROUTES = frozenset({"no_action"})
 STRATEGY_PLUGIN_ALERT_ACTIONS = frozenset({"defend", "blocked"})
+STRATEGY_PLUGIN_AUTOMATED_POSITION_ACTIONS = frozenset({"defend", "delever"})
 CRISIS_RESPONSE_SHADOW_SUPPORTED_STRATEGIES = frozenset(
     {
         "tqqq_growth_income",
@@ -930,13 +931,37 @@ def build_strategy_plugin_notification_lines(
 def should_alert_strategy_plugin_signal(signal: StrategyPluginSignal) -> bool:
     route = _normalize_strategy_plugin_field(getattr(signal, "canonical_route", None))
     action = _normalize_strategy_plugin_field(getattr(signal, "suggested_action", None))
+    if _is_strategy_position_control_notice(signal, action=action):
+        return False
     return (
         bool(getattr(signal, "would_trade_if_enabled", False))
         or route not in STRATEGY_PLUGIN_NON_ALERT_ROUTES
         or action in STRATEGY_PLUGIN_ALERT_ACTIONS
     )
 
 
+def _is_strategy_position_control_notice(signal: StrategyPluginSignal, *, action: str | None = None) -> bool:
+    """Return true when strategy runtime should carry the alert instead of the plugin bot."""
+
+    target_type = _normalize_strategy_plugin_field(getattr(signal, "target_type", None)) or "strategy"
+    if target_type != "strategy":
+        return False
+    normalized_action = action if action is not None else _normalize_strategy_plugin_field(
+        getattr(signal, "suggested_action", None)
+    )
+    if normalized_action not in STRATEGY_PLUGIN_AUTOMATED_POSITION_ACTIONS:
+        return False
+    controls = getattr(signal, "execution_controls", {}) or {}
+    if not isinstance(controls, Mapping):
+        return False
+    if not _as_bool(controls.get("strategy_runtime_metadata_allowed"), default=False):
+        return False
+    if not _as_bool(controls.get("position_control_allowed"), default=False):
+        return False
+    evidence_status = _normalize_strategy_plugin_field(controls.get("consumption_evidence_status"))
+    return evidence_status == "automation_approved"
+
+
 def build_strategy_plugin_alert_guidance(
     signal: StrategyPluginSignal,
     *,

tests/test_strategy_plugins.py

@@ -663,6 +663,71 @@ def test_strategy_plugin_no_action_signal_does_not_escalate_alert(self):
         self.assertFalse(should_alert_strategy_plugin_signal(signal))
         self.assertEqual(build_strategy_plugin_alert_messages([signal]), ())
 
+    def test_strategy_plugin_auto_position_control_signal_stays_with_strategy_notification(self):
+        signal = validate_strategy_plugin_signal_payload(
+            {
+                **_signal_payload(plugin=PLUGIN_MARKET_REGIME_CONTROL),
+                "canonical_route": "risk_off",
+                "suggested_action": "defend",
+                "would_trade_if_enabled": True,
+                "execution_controls": {
+                    **_signal_payload()["execution_controls"],
+                    "strategy_runtime_metadata_allowed": True,
+                    "position_control_allowed": True,
+                    "consumption_evidence_status": "automation_approved",
+                },
+            }
+        )
+
+        self.assertFalse(should_alert_strategy_plugin_signal(signal))
+        self.assertEqual(build_strategy_plugin_alert_messages([signal]), ())
+
+    def test_strategy_plugin_notification_target_still_alerts_plugin_bot(self):
+        signal = validate_strategy_plugin_signal_payload(
+            {
+                **_signal_payload(plugin=PLUGIN_MARKET_REGIME_CONTROL),
+                "target_type": "notification_target",
+                "strategy": "",
+                "notification_target": GENERAL_MARKET_REGIME_NOTIFICATION_TARGET,
+                "canonical_route": "risk_off",
+                "suggested_action": "defend",
+                "would_trade_if_enabled": True,
+                "execution_controls": {
+                    **_signal_payload()["execution_controls"],
+                    "strategy_runtime_metadata_allowed": False,
+                    "position_control_allowed": False,
+                    "consumption_evidence_status": "notification_only",
+                    "capital_impact": "notification_only",
+                },
+            }
+        )
+
+        self.assertTrue(should_alert_strategy_plugin_signal(signal))
+        alerts = build_strategy_plugin_alert_messages([signal])
+        self.assertEqual(len(alerts), 1)
+        self.assertEqual(alerts[0].metadata["target_type"], "notification_target")
+
+    def test_strategy_plugin_manual_review_strategy_signal_still_alerts_plugin_bot(self):
+        signal = validate_strategy_plugin_signal_payload(
+            {
+                **_signal_payload(plugin=PLUGIN_MARKET_REGIME_CONTROL),
+                "canonical_route": "opportunity_watch",
+                "suggested_action": "notify_manual_review",
+                "would_trade_if_enabled": False,
+                "execution_controls": {
+                    **_signal_payload()["execution_controls"],
+                    "strategy_runtime_metadata_allowed": True,
+                    "position_control_allowed": True,
+                    "consumption_evidence_status": "automation_approved",
+                },
+            }
+        )
+
+        self.assertTrue(should_alert_strategy_plugin_signal(signal))
+        alerts = build_strategy_plugin_alert_messages([signal])
+        self.assertEqual(len(alerts), 1)
+        self.assertIn("Manual review only", alerts[0].body)
+
     def test_strategy_plugin_true_crisis_builds_generic_alert_message(self):
         signal = validate_strategy_plugin_signal_payload(
             {