From 861b4f540e444f2e3fbf337f6232d829b6a32cc6 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 5 Nov 2026 05:18:04 +0800 Subject: [PATCH] fix: Unify Firstrade operational error notifications --- application/rebalance_service.py | 22 ++- main.py | 16 +- notifications/unified.py | 72 ++++++++- scripts/cloud_run_runtime_guard.py | 15 +- scripts/execution_report_heartbeat.py | 71 ++++---- tests/fixtures/notification_p10_errors.json | 170 ++++++++++++++++++++ tests/test_account_alert_redaction.py | 5 +- tests/test_cloud_run_runtime_guard.py | 6 +- tests/test_notification_error_format.py | 83 ++++++++++ tests/test_notification_off_equivalence.py | 18 +++ tests/test_notification_quiet.py | 5 +- tests/test_request_handling.py | 13 +- tests/test_runtime_notification_i18n.py | 11 +- 13 files changed, 421 insertions(+), 86 deletions(-) create mode 100644 tests/fixtures/notification_p10_errors.json create mode 100644 tests/test_notification_error_format.py diff --git a/application/rebalance_service.py b/application/rebalance_service.py index cf1bfed0..1f72c118 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -275,13 +275,33 @@ def _publish_cycle_notification( settings: PlatformRuntimeSettings, notification_sender: Callable[[str], None] | None = None, log_message: Callable[[str], None] = print, + report: Mapping[str, Any] | None = None, ) -> bool: sender = notification_sender if sender is None: if not settings.tg_token or not settings.tg_chat_id: return False sender = build_sender(settings.tg_token, settings.tg_chat_id) - notification = render_cycle_notification(result, lang=settings.notify_lang) + from notifications.unified import alert_text, cycle_kind + kind = cycle_kind(result) + if kind in {"routine_cycle", "order_submitted", "fill"}: + notification = render_cycle_notification(result, lang=settings.notify_lang) + else: + from quant_platform_kit.notifications.events import RenderedNotification + from datetime import datetime + observed_at = None + if report: + try: + observed_at = datetime.fromisoformat(str(report.get("started_at", "")).replace("Z", "+00:00")) + except ValueError: + pass + skips = result.get("execution_blocking_skips") or result.get("skipped_orders") or () + display_kind = "fractional_trading_disclosure_required" if any( + isinstance(item, Mapping) and item.get("reason") == "fractional_trading_disclosure_required" + for item in skips) else kind + text = alert_text(display_kind, settings.notify_lang, occurred_at=observed_at, + run_url=(report or {}).get("workflow_run_url")) + notification = RenderedNotification(detailed_text=text, compact_text=text) def publish_log(text: str) -> None: try: log_message(text, flush=True) diff --git a/main.py b/main.py index 5197b467..7d8f6612 100644 --- a/main.py +++ b/main.py @@ -196,16 +196,10 @@ def _telegram_notification_targets() -> tuple[tuple[str, str], ...]: def _runtime_error_notification_message(exc: Exception) -> str: - t = build_translator(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG")) - title = "runtime_probe_failure_title" if request.path == "/probe" else "runtime_failure_title" - strategy_name = os.getenv("STRATEGY_DISPLAY_NAME") or os.getenv("STRATEGY_PROFILE") or "Firstrade" - return "\n".join(( - t(title), - t("strategy_label", name=strategy_name), - t("runtime_failure_context", context=os.getenv('ACCOUNT_REGION') or os.getenv('K_SERVICE') or 'Firstrade'), - t("runtime_failure_result"), - t("runtime_failure_action"), - )) + from notifications.unified import alert_text, coverage_locale, workflow_run_url + return alert_text("real_error", coverage_locale(), occurred_at=datetime.now(timezone.utc), + source="heartbeat" if request.path == "/probe" else "runtime", + run_url=workflow_run_url()) def _notify_runtime_error(exc: Exception) -> bool: @@ -550,7 +544,7 @@ def send_failure(): return result result["report_persistence_error"] = "execution_report_write_unconfirmed" delivery = notify_cycle(report, result, settings=settings, - send=lambda: _publish_cycle_notification(result, settings=settings), + send=lambda: _publish_cycle_notification(result, settings=settings, report=report), send_failure=send_failure) result["notification_consolidation"] = delivery result["notification_sent"] = delivery.get("transport_acknowledged") is True diff --git a/notifications/unified.py b/notifications/unified.py index 28af849d..d4e1548c 100644 --- a/notifications/unified.py +++ b/notifications/unified.py @@ -606,13 +606,69 @@ def notify_heartbeat(*, kind, send, now, env=None, store=None, recovery_reports= return {"candidate_persisted": False, "failed": True} -def alert_text(kind, locale): +def alert_text(kind, locale=None, *, occurred_at=None, source="runtime", run_url=None): + """Plan P10 presentation contract; never interpolate private runtime payloads. + + Monitor observations are not business runs. Missing source time/link stays + unverified. Replace this renderer with the step-1 shared formatter once its + reviewed dependency pin is available; no vendor snapshot is edited here. + """ + from quant_platform_kit.notifications._redaction import redact_sensitive_text + from zoneinfo import ZoneInfo + + zh = not str(locale or "zh").strip().lower().startswith("en") messages = { - "receipt_missing": ("执行回执缺失。", "Execution receipt is missing."), - "receipt_unacceptable": ("执行回执需要检查。", "Execution receipt needs review."), - "real_error": ("运行发生异常。", "The runtime encountered an error."), - "candidate_write_failure": ("通知候选未能确认写入。", "Notification candidate storage could not be confirmed."), + "receipt_missing": ("执行回执缺失", "Execution receipt is missing"), + "receipt_unacceptable": ("执行回执无法核实", "Execution receipt cannot be verified"), + "receipt_identity_conflict": ("执行回执身份冲突", "Execution receipt identity conflict"), + "receipt_stale": ("执行回执早于应运行时间", "Execution receipt predates the required run"), + "real_error": ("运行发生异常", "Runtime error"), + "auth_failure": ("认证失败", "Authentication failed"), + "blocked_run": ("执行被阻断", "Execution is blocked"), + "risk_escalation": ("风险检查需要处理", "Risk check requires attention"), + "order_status_unknown": ("订单结果无法核实", "Order outcome is unknown"), + "candidate_write_failure": ("通知候选未能确认写入", "Notification candidate storage could not be confirmed"), + "runtime_guard": ("运行监控发现异常", "Runtime monitor detected an error"), } - zh = str(locale or "").strip().lower().startswith("zh") - title = messages.get(kind, messages["receipt_unacceptable"])[0 if zh else 1] - return title + ("\n请查看本次运行记录。" if zh else "\nReview the current run record.") + problem = messages.get(kind, messages["receipt_unacceptable"])[0 if zh else 1] + if kind == "fractional_trading_disclosure_required": + from notifications.telegram import build_translator + problem = build_translator("zh" if zh else "en")("skip_reason_fractional_trading_disclosure_required") + stamp = "未核实" if zh else "unverified" + if isinstance(occurred_at, datetime) and occurred_at.tzinfo is not None and occurred_at.utcoffset() is not None: + stamp = occurred_at.astimezone(ZoneInfo("Asia/Singapore")).strftime("%Y-%m-%d %H:%M:%S") + if not zh: + stamp += " SGT" + else: + stamp += "（新加坡时间）" + producers = { + "runtime": ("策略运行", "strategy runtime"), + "heartbeat": ("回执检查", "receipt monitor"), + "runtime_guard": ("运行监控", "runtime monitor"), + } + producer = producers.get(source, producers["runtime"])[0 if zh else 1] + # Only authentic first-party Actions references, never arbitrary hostnames, + # query strings, account selectors or cloud resource paths. + link = "运行链接不可用" if zh else "run link unavailable" + if isinstance(run_url, str) and re.fullmatch( + r"https://github\.com/QuantStrategyLab/FirstradePlatform/actions/runs/[0-9]+", run_url + ): + link = redact_sensitive_text(run_url) + if zh: + lines = ["【资产管理｜运行错误】", f"时间：{stamp}", "平台 / 策略：第一证券 / 未核实", + f"问题：{problem}", "影响：是否提交订单：未知；是否有待处理订单：未知", + f"证据：{producer}；{link}", "处理：查看本次运行记录，核实输入；未恢复前保持执行保护", + "事件：未提供安全事件引用｜投递状态以回执为准"] + else: + lines = ["[Asset management | Runtime error]", f"Time: {stamp}", "Platform / Strategy: Firstrade / unverified", + f"Problem: {problem}", "Impact: order submission unknown; pending orders unknown", + f"Evidence: {producer}; {link}", "Action: review this run and its inputs; retain execution protection until recovery", + "Event: safe reference unavailable | Delivery status follows the receipt"] + return redact_sensitive_text("\n".join(lines)) + + +def workflow_run_url(env=None): + """Return the real monitor workflow reference, never a synthesized business run.""" + env = os.environ if env is None else env + return (f"{env.get('GITHUB_SERVER_URL', '')}/{env.get('GITHUB_REPOSITORY', '')}" + f"/actions/runs/{env.get('GITHUB_RUN_ID', '')}") diff --git a/scripts/cloud_run_runtime_guard.py b/scripts/cloud_run_runtime_guard.py index 04b6f9c3..5a74ac5f 100644 --- a/scripts/cloud_run_runtime_guard.py +++ b/scripts/cloud_run_runtime_guard.py @@ -72,7 +72,7 @@ def _env_bool(name: str, default: bool = False) -> bool: def _notification_locale() -> str: - return resolve_operational_notification_locale(os.environ.get("NOTIFY_LANG")) + return resolve_operational_notification_locale(os.environ.get("QSL_NOTIFY_LANG") or os.environ.get("NOTIFY_LANG")) def _notice(key: str, /, **values: object) -> str: @@ -792,16 +792,9 @@ def main() -> int: f"{os.environ['GITHUB_SERVER_URL']}/{os.environ['GITHUB_REPOSITORY']}" f"/actions/runs/{os.environ['GITHUB_RUN_ID']}" ) - message = format_operational_alert( - locale=_notification_locale(), - alert_type="runtime_guard", - name=name, - context={"project": project, "lookback_minutes": lookback_minutes}, - issues=issues, - # Keep provider audit payloads in their source logs, including on pinned QPK. - technical_details=(), - workflow_url=run_url, - ) + from notifications.unified import alert_text + message = alert_text("runtime_guard", _notification_locale(), occurred_at=dt.datetime.now(dt.timezone.utc), + source="runtime_guard", run_url=run_url) print(message) decision = alert_dedupe.decide(name=name, issues=issues, event_ids=alert_event_ids) print(decision.log_line()) diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index af515a8c..a0f83982 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -55,7 +55,7 @@ def _env_bool(name: str, default: bool = False) -> bool: def _notification_locale() -> str: - return resolve_operational_notification_locale(os.environ.get("NOTIFY_LANG")) + return resolve_operational_notification_locale(os.environ.get("QSL_NOTIFY_LANG") or os.environ.get("NOTIFY_LANG")) def _notice(key: str, /, **values: object) -> str: @@ -760,52 +760,45 @@ def _send_telegram(message: str) -> bool: def _deliver_alert(message: str, *, kind: str, now: dt.datetime, affected_keys=None) -> bool: - from notifications.unified import alert_text, enabled, notify_heartbeat + from notifications.unified import alert_text, enabled, notify_heartbeat, workflow_run_url + display_kind = "receipt_identity_conflict" if "conflicting report identity" in message else kind + message = alert_text(display_kind, _notification_locale(), occurred_at=now, + source="heartbeat", run_url=workflow_run_url()) if not enabled(): return _send_telegram(message) - result = notify_heartbeat(kind=kind, send=lambda: _send_telegram(alert_text(kind, _notification_locale())), + result = notify_heartbeat(kind=kind, send=lambda: _send_telegram(message), now=now, affected_keys=affected_keys) return any(item.get("transport_acknowledged") for item in result.get("deliveries", [])) def _send_normal_heartbeat(name: str, detail: str, *, quiet_eligible: bool = True, now: dt.datetime | None = None, recovery_reports=(), affected_keys=None) -> int: - from notifications.unified import alert_text, coverage_locale, enabled, notify_heartbeat - if quiet_eligible or enabled(): - locale = coverage_locale() if quiet_eligible else _notification_locale() - message = alert_text("candidate_write_failure" if quiet_eligible else "receipt_unacceptable", locale) - def send_attention(): - marker = os.environ.get("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH") - if quiet_eligible and marker: - try: - Path(marker).write_text("attempted\n") - except OSError: - pass - acknowledged = _send_telegram(message[:3900]) - if quiet_eligible and acknowledged is True and marker: - try: - Path(marker).write_text("acknowledged\n") - except OSError: - pass - return acknowledged - result = notify_heartbeat(kind="healthy_heartbeat" if quiet_eligible else "receipt_unacceptable", - send=send_attention, now=now or dt.datetime.now(dt.timezone.utc), - recovery_reports=recovery_reports, affected_keys=affected_keys) - return 1 if result.get("failed") or not result.get("candidate_persisted") else 0 - if quiet_eligible and not _env_bool("RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS", False): - return 0 - if quiet_eligible: - message = format_operational_heartbeat_status( - locale=_notification_locale(), name=name, detail=detail, - ) - else: - message = format_operational_alert( - locale=_notification_locale(), alert_type="execution_report_heartbeat", name=name, - issues=["运行检查证据待核对" if _notification_locale() == "zh" else "Runtime check evidence needs review"], - technical_details=[detail], - ) - print(message) - return 0 if _send_telegram(message[:3900]) else 1 + from notifications.unified import alert_text, coverage_locale, enabled, notify_heartbeat, workflow_run_url + locale = coverage_locale() if quiet_eligible else _notification_locale() + kind = "candidate_write_failure" if quiet_eligible else ( + "receipt_stale" if "predates latest due" in detail else "receipt_unacceptable") + message = alert_text(kind, locale, + occurred_at=now, source="heartbeat", run_url=workflow_run_url()) + if not quiet_eligible and not enabled(): + return 0 if _send_telegram(message) else 1 + def send_attention(): + marker = os.environ.get("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH") + if quiet_eligible and marker: + try: + Path(marker).write_text("attempted\n") + except OSError: + pass + acknowledged = _send_telegram(message[:3900]) + if quiet_eligible and acknowledged is True and marker: + try: + Path(marker).write_text("acknowledged\n") + except OSError: + pass + return acknowledged + result = notify_heartbeat(kind="healthy_heartbeat" if quiet_eligible else "receipt_unacceptable", + send=send_attention, now=now or dt.datetime.now(dt.timezone.utc), + recovery_reports=recovery_reports, affected_keys=affected_keys) + return 1 if result.get("failed") or not result.get("candidate_persisted") else 0 def main(now: dt.datetime | None = None) -> int: diff --git a/tests/fixtures/notification_p10_errors.json b/tests/fixtures/notification_p10_errors.json new file mode 100644 index 00000000..b33aad65 --- /dev/null +++ b/tests/fixtures/notification_p10_errors.json @@ -0,0 +1,170 @@ +{ + "heartbeat:zh:blocked:False": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "read", + "gs://synthetic/reports/source.json", + { + "project": null + } + ], + [ + "send", + "【资产管理｜运行错误】\n时间：2026-10-17 07:00:00（新加坡时间）\n平台 / 策略：第一证券 / 未核实\n问题：执行回执无法核实\n影响：是否提交订单：未知；是否有待处理订单：未知\n证据：回执检查；运行链接不可用\n处理：查看本次运行记录，核实输入；未恢复前保持执行保护\n事件：未提供安全事件引用｜投递状态以回执为准" + ] + ] + }, + "heartbeat:zh:blocked:True": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "read", + "gs://synthetic/reports/source.json", + { + "project": null + } + ], + [ + "send", + "【资产管理｜运行错误】\n时间：2026-10-17 07:00:00（新加坡时间）\n平台 / 策略：第一证券 / 未核实\n问题：执行回执无法核实\n影响：是否提交订单：未知；是否有待处理订单：未知\n证据：回执检查；运行链接不可用\n处理：查看本次运行记录，核实输入；未恢复前保持执行保护\n事件：未提供安全事件引用｜投递状态以回执为准" + ] + ] + }, + "heartbeat:zh:missing:False": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "send", + "【资产管理｜运行错误】\n时间：2026-10-17 07:00:00（新加坡时间）\n平台 / 策略：第一证券 / 未核实\n问题：执行回执缺失\n影响：是否提交订单：未知；是否有待处理订单：未知\n证据：回执检查；运行链接不可用\n处理：查看本次运行记录，核实输入；未恢复前保持执行保护\n事件：未提供安全事件引用｜投递状态以回执为准" + ] + ] + }, + "heartbeat:zh:missing:True": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "send", + "【资产管理｜运行错误】\n时间：2026-10-17 07:00:00（新加坡时间）\n平台 / 策略：第一证券 / 未核实\n问题：执行回执缺失\n影响：是否提交订单：未知；是否有待处理订单：未知\n证据：回执检查；运行链接不可用\n处理：查看本次运行记录，核实输入；未恢复前保持执行保护\n事件：未提供安全事件引用｜投递状态以回执为准" + ] + ] + }, + "runtime_error:zh": { + "result": true, + "sends": [ + "【资产管理｜运行错误】\n时间：2026-10-17 07:00:00（新加坡时间）\n平台 / 策略：第一证券 / 未核实\n问题：运行发生异常\n影响：是否提交订单：未知；是否有待处理订单：未知\n证据：策略运行；运行链接不可用\n处理：查看本次运行记录，核实输入；未恢复前保持执行保护\n事件：未提供安全事件引用｜投递状态以回执为准" + ] + }, + "heartbeat:en:blocked:False": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "read", + "gs://synthetic/reports/source.json", + { + "project": null + } + ], + [ + "send", + "[Asset management | Runtime error]\nTime: 2026-10-17 07:00:00 SGT\nPlatform / Strategy: Firstrade / unverified\nProblem: Execution receipt cannot be verified\nImpact: order submission unknown; pending orders unknown\nEvidence: receipt monitor; run link unavailable\nAction: review this run and its inputs; retain execution protection until recovery\nEvent: safe reference unavailable | Delivery status follows the receipt" + ] + ] + }, + "heartbeat:en:blocked:True": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "read", + "gs://synthetic/reports/source.json", + { + "project": null + } + ], + [ + "send", + "[Asset management | Runtime error]\nTime: 2026-10-17 07:00:00 SGT\nPlatform / Strategy: Firstrade / unverified\nProblem: Execution receipt cannot be verified\nImpact: order submission unknown; pending orders unknown\nEvidence: receipt monitor; run link unavailable\nAction: review this run and its inputs; retain execution protection until recovery\nEvent: safe reference unavailable | Delivery status follows the receipt" + ] + ] + }, + "heartbeat:en:missing:False": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "send", + "[Asset management | Runtime error]\nTime: 2026-10-17 07:00:00 SGT\nPlatform / Strategy: Firstrade / unverified\nProblem: Execution receipt is missing\nImpact: order submission unknown; pending orders unknown\nEvidence: receipt monitor; run link unavailable\nAction: review this run and its inputs; retain execution protection until recovery\nEvent: safe reference unavailable | Delivery status follows the receipt" + ] + ] + }, + "heartbeat:en:missing:True": { + "exit_code": 1, + "calls": [ + [ + "list", + "gs://synthetic/reports", + { + "project": null + } + ], + [ + "send", + "[Asset management | Runtime error]\nTime: 2026-10-17 07:00:00 SGT\nPlatform / Strategy: Firstrade / unverified\nProblem: Execution receipt is missing\nImpact: order submission unknown; pending orders unknown\nEvidence: receipt monitor; run link unavailable\nAction: review this run and its inputs; retain execution protection until recovery\nEvent: safe reference unavailable | Delivery status follows the receipt" + ] + ] + }, + "runtime_error:en": { + "result": true, + "sends": [ + "[Asset management | Runtime error]\nTime: 2026-10-17 07:00:00 SGT\nPlatform / Strategy: Firstrade / unverified\nProblem: Runtime error\nImpact: order submission unknown; pending orders unknown\nEvidence: strategy runtime; run link unavailable\nAction: review this run and its inputs; retain execution protection until recovery\nEvent: safe reference unavailable | Delivery status follows the receipt" + ] + } +} diff --git a/tests/test_account_alert_redaction.py b/tests/test_account_alert_redaction.py index 9baacda6..077b03bd 100644 --- a/tests/test_account_alert_redaction.py +++ b/tests/test_account_alert_redaction.py @@ -77,7 +77,7 @@ def post(url, *, json, timeout): assert "******21" in payload["error"] assert len(sent) == 1 assert sent[0]["chat_id"] == "synthetic-chat" - assert ("health check failed" if lang == "en" else "健康检查失败") in sent[0]["text"] + assert sent[0]["text"].startswith("[Asset management | Runtime error]" if lang == "en" else "【资产管理｜运行错误】") assert "Requested account" not in sent[0]["text"] surfaces = [response.get_data(as_text=True), json.dumps(sent), capsys.readouterr().out] for surface in surfaces: @@ -107,7 +107,8 @@ def post(url, *, json, timeout): settings = SimpleNamespace(tg_token="synthetic-token", tg_chat_id="synthetic-chat", notify_lang=lang) assert _publish_cycle_notification(result, settings=settings, log_message=logged.append) is True assert len(sent) == 1 - assert logged and "******21" in "\n".join(logged) + assert logged and "******21" not in "\n".join(logged) + assert ("Execution is blocked" if lang == "en" else "执行被阻断") in "\n".join(logged) for surface in (json.dumps(sent), "\n".join(logged)): assert "87654321" not in surface assert "4321" not in surface diff --git a/tests/test_cloud_run_runtime_guard.py b/tests/test_cloud_run_runtime_guard.py index 1094697b..6dcad53d 100644 --- a/tests/test_cloud_run_runtime_guard.py +++ b/tests/test_cloud_run_runtime_guard.py @@ -349,8 +349,10 @@ def test_runtime_guard_keeps_provider_audit_payload_out_of_notification(monkeypa assert guard.main() == 1 assert len(messages) == 1 - assert "失败日志" in messages[0] - assert "https://example.test/example/runtime/actions/runs/1" in messages[0] + assert messages[0].startswith("【资产管理｜运行错误】") + assert "运行监控发现异常" in messages[0] + assert "运行链接不可用" in messages[0] + assert "https://example.test" not in messages[0] assert "AuditLog" not in messages[0] assert "principalEmail" not in messages[0] assert "example@example.test" not in messages[0] diff --git a/tests/test_notification_error_format.py b/tests/test_notification_error_format.py new file mode 100644 index 00000000..f2873a7b --- /dev/null +++ b/tests/test_notification_error_format.py @@ -0,0 +1,83 @@ +"""Offline P10 error format regressions; synthetic inputs only.""" +from datetime import datetime, timezone +from types import SimpleNamespace + +import pytest + +from notifications import unified +from test_notification_consolidation import ENV, Store, candidate, source, NOW + + +@pytest.mark.parametrize("flag", ["true", "false"]) +@pytest.mark.parametrize("changes", [ + {"ok": False, "error": "HTTP 503"}, + {"execution_blocked": True}, + {"orders_pending_count": 1}, + {"report_persistence_error": "missing"}, +]) +def test_cycle_errors_use_fixed_template_without_healthy_claims(flag, changes): + from application.rebalance_service import _publish_cycle_notification + sent = [] + report, result = source(**changes) + result.update(account="fake-private-account", strategy_display_name="fake-private-alias") + settings = SimpleNamespace(notify_lang="zh") + outcome = unified.notify_cycle(report, result, settings=settings, + send=lambda: _publish_cycle_notification(result, settings=settings, + notification_sender=lambda text: sent.append(text) or True, log_message=lambda *a, **kw: None), + env={**ENV, unified.FLAG: flag}, now=NOW, store=Store()) + assert outcome["transport_acknowledged"] is True + assert len(sent) == 1 + assert sent[0].startswith("【资产管理｜运行错误】") + assert "是否提交订单：未知" in sent[0] + assert not any(value in sent[0] for value in ("无需调仓", "心跳", "fake-private", "503")) + + +@pytest.mark.parametrize("locale", ["zh", "en"]) +def test_error_template_preserves_unknown_time_and_evidence(locale): + text = unified.alert_text("fake-private-kind", locale) + assert text.startswith("【资产管理｜运行错误】" if locale == "zh" else "[Asset management | Runtime error]") + assert "fake-private-kind" not in text + assert ("时间：未核实" if locale == "zh" else "Time: unverified") in text + assert ("运行链接不可用" if locale == "zh" else "run link unavailable") in text + + +def test_heartbeat_flag_off_uses_same_error_format(monkeypatch): + from scripts import execution_report_heartbeat as heartbeat + monkeypatch.setenv(unified.FLAG, "false") + monkeypatch.setenv("QSL_NOTIFY_LANG", "zh") + sent = [] + monkeypatch.setattr(heartbeat, "_send_telegram", lambda text: sent.append(text) or True) + assert heartbeat._deliver_alert("fake-private-alias HTTP 503", kind="receipt_missing", now=NOW) + assert sent[0].startswith("【资产管理｜运行错误】") + assert "fake-private" not in sent[0] + + +@pytest.mark.parametrize("flag", ["true", "false"]) +def test_healthy_cycle_writes_candidate_without_legacy_body(flag): + report, result = source() + store = Store() + def forbidden(): + pytest.fail("healthy result must not send an immediate body") + outcome = unified.notify_cycle(report, result, settings=None, send=forbidden, + env={**ENV, unified.FLAG: flag}, now=NOW, store=store) + assert outcome["candidate_persisted"] is True + assert candidate(store)["runs"][0]["cycle_count"] == 1 + + +@pytest.mark.parametrize("url", [ + "https://example.test/private-account/actions/runs/1", + "https://github.com/QuantStrategyLab/FirstradePlatform/actions/runs/1?token=fake-secret", + "https://user:fake-secret@github.com/QuantStrategyLab/FirstradePlatform/actions/runs/1", +]) +def test_private_or_untrusted_evidence_url_is_not_echoed(url): + text = unified.alert_text("real_error", "en", run_url=url) + assert url not in text and "fake-secret" not in text + assert "run link unavailable" in text + + +def test_authentic_workflow_link_and_observed_time_are_preserved(): + url = "https://github.com/QuantStrategyLab/FirstradePlatform/actions/runs/1" + text = unified.alert_text("runtime_guard", "en", occurred_at=NOW, + source="runtime_guard", run_url=url) + assert url in text and "2026-10-17 08:00:00 SGT" in text + assert "runtime monitor" in text diff --git a/tests/test_notification_off_equivalence.py b/tests/test_notification_off_equivalence.py index e4e2fa28..0254ba9c 100644 --- a/tests/test_notification_off_equivalence.py +++ b/tests/test_notification_off_equivalence.py @@ -13,10 +13,22 @@ import main from application.rebalance_service import _publish_cycle_notification from notifications.telegram import render_cycle_notification +from notifications import unified from scripts import execution_report_heartbeat as heartbeat GOLDEN = Path(__file__).parent / "fixtures" / "notification_main_c320fa0.json" NOW = datetime(2026, 10, 16, 23, tzinfo=timezone.utc) +ERROR_GOLDEN = Path(__file__).parent / "fixtures" / "notification_p10_errors.json" + + +class FixedDateTime(datetime): + @classmethod + def now(cls, tz=None): + return NOW.astimezone(tz) if tz else NOW.replace(tzinfo=None) + + +def missing_candidate_store(): + raise RuntimeError("synthetic_candidate_store_unavailable") def record_main(monkeypatch, *, outcome, lang, notify=True): @@ -160,8 +172,14 @@ def test_flags_off_entrypoints_renderers_match_main_golden(monkeypatch, flag, mo else: monkeypatch.setenv("FIRSTRADE_UNIFIED_NOTIFICATION_ENABLED", flag) monkeypatch.setenv("QSL_ROUTINE_NOTIFICATION_MODE", mode) + # Offline: the fixture intentionally lacks a source-bound candidate store. + # Do not let the SDK attempt metadata/credential discovery. + monkeypatch.setattr(unified, "GcsJsonStore", missing_candidate_store) + monkeypatch.setattr(main, "datetime", FixedDateTime) actual = record_all(monkeypatch) expected = json.loads(GOLDEN.read_text())[mode] + # Retain the historical broker/state golden; P10 changes only error bodies. + expected.update(json.loads(ERROR_GOLDEN.read_text())) for lang in ("zh", "en"): for outcome in ("no_action", "filled", "error"): key = f"main:{lang}:{outcome}" diff --git a/tests/test_notification_quiet.py b/tests/test_notification_quiet.py index 555f95fc..9672bd1b 100644 --- a/tests/test_notification_quiet.py +++ b/tests/test_notification_quiet.py @@ -170,7 +170,8 @@ 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 'Execution receipt identity conflict' in sent[0] + assert 'wrong-scope' not in sent[0] @pytest.mark.parametrize('value', [True, 1, 'bad']) @@ -337,4 +338,4 @@ 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 'Execution receipt predates the required run' in sent[0] diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index a0914468..772119a3 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -455,6 +455,7 @@ def test_probe_endpoint_calls_service_when_gate_enabled(monkeypatch): def test_probe_endpoint_notifies_only_on_error(monkeypatch): + monkeypatch.setenv("QSL_NOTIFY_LANG", "en") monkeypatch.setenv("FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP", "true") monkeypatch.setenv("TELEGRAM_TOKEN", "token-1") monkeypatch.setenv("GLOBAL_TELEGRAM_CHAT_ID", "chat-1") @@ -483,11 +484,13 @@ def send(message): assert len(sent_messages) == 1 assert sent_messages[0][0] == "token-1" assert sent_messages[0][1] == "chat-1" - assert "Firstrade health check failed" in sent_messages[0][2] + assert sent_messages[0][2].startswith("[Asset management | Runtime error]") + assert "receipt monitor" in sent_messages[0][2] assert "RuntimeError: session denied" not in sent_messages[0][2] def test_run_endpoint_notifies_telegram_on_strategy_cycle_error(monkeypatch): + monkeypatch.setenv("QSL_NOTIFY_LANG", "en") sent_messages = [] def fake_build_sender(token, chat_id): @@ -521,9 +524,9 @@ def send(message): assert len(sent_messages) == 1 assert sent_messages[0][0] == "token-1" assert sent_messages[0][1] == "chat-1" - assert "Firstrade strategy run failed" in sent_messages[0][2] + assert "[Asset management | Runtime error]" in sent_messages[0][2] assert "ValueError: snapshot denied" not in sent_messages[0][2] - assert "Strategy: russell_top50_leader_rotation" in sent_messages[0][2] + assert "Platform / Strategy: Firstrade / unverified" in sent_messages[0][2] def test_run_endpoint_error_notification_uses_chinese_copy(monkeypatch): @@ -553,8 +556,8 @@ def send(message): assert response.status_code == 500 text = sent_messages[0][2] - assert "Firstrade 策略运行失败" in text - assert "策略: russell_top50_leader_rotation" in text + assert text.startswith("【资产管理｜运行错误】") + assert "平台 / 策略：第一证券 / 未核实" in text assert "snapshot denied" not in text diff --git a/tests/test_runtime_notification_i18n.py b/tests/test_runtime_notification_i18n.py index 8b46bd55..6f6efef2 100644 --- a/tests/test_runtime_notification_i18n.py +++ b/tests/test_runtime_notification_i18n.py @@ -1,4 +1,5 @@ import ast +from datetime import datetime, timezone import os from pathlib import Path from types import SimpleNamespace @@ -16,7 +17,7 @@ def test_chinese_locale_variants_use_chinese(locale): def runtime_function(name, **overrides): tree = ast.parse((Path(__file__).resolve().parents[1] / "main.py").read_text()) function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name) - namespace = dict(os=os, request=SimpleNamespace(method="GET", path="/main"), + namespace = dict(os=os, datetime=datetime, timezone=timezone, request=SimpleNamespace(method="GET", path="/main"), NOTIFY_LANG="zh-CN", SERVICE_NAME="Test service", SECRET_NAME="Test secret alias", STRATEGY_PROFILE="test_strategy", ACCOUNT_REGION="SG", ACCOUNT_GROUP="Test account", strategy_display_name="Test strategy", build_translator=build_translator, @@ -34,13 +35,13 @@ def test_startup_alert_is_compact_localized_and_does_not_embed_provider_text(loc monkeypatch.setenv("STRATEGY_PROFILE", "test_strategy") fn = runtime_function("_runtime_error_notification_message", NOTIFY_LANG=locale, t=build_translator(locale)) message = fn(RuntimeError("PRIVATE_PROVIDER_SENTINEL")) - assert len(message.splitlines()) <= 5 + assert len(message.splitlines()) == 8 assert "PRIVATE_PROVIDER_SENTINEL" not in message assert "未提交订单" not in message assert "no order" not in message.lower() - assert ("未正常结束" in message) is locale.startswith("zh") - assert ("did not finish successfully" in message) is (locale == "en") - assert "test_strategy" in message or "Test strategy" in message + assert message.startswith("【资产管理｜运行错误】" if locale.startswith("zh") else "[Asset management | Runtime error]") + assert "test_strategy" not in message and "Test strategy" not in message + assert ("是否提交订单：未知" if locale.startswith("zh") else "order submission unknown") in message def test_failed_fallback_delivery_logs_failure_and_preserves_attempted_semantics(capsys):