From 1c8b77a30af2fb7c7d642ec4f39650fac3575fec Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:47:33 +0800 Subject: [PATCH 1/2] fix: harden runtime lifecycle and dispatch Co-Authored-By: Codex --- .github/workflows/main.yml | 29 +++-- application/cycle_service.py | 9 +- docs/operator_runbook.md | 8 +- live_services.py | 31 ++++- main.py | 8 +- reporting/status_reports.py | 10 +- runtime_config_support.py | 61 +++++++++- runtime_support.py | 109 ++++++++++++++++-- scripts/runtime_workflow_heartbeat.py | 11 +- tests/test_live_services.py | 31 +++++ tests/test_main_runtime_error_notification.py | 19 +++ tests/test_runtime_config_support.py | 48 ++++++++ tests/test_runtime_support.py | 55 ++++++++- tests/test_runtime_trigger_workflow.py | 12 ++ tests/test_runtime_workflow_heartbeat.py | 21 ++++ tests/test_runtime_workflow_shared_config.sh | 2 + tests/test_status_reports.py | 44 +++++++ tests/test_strategy_runtime.py | 5 +- 18 files changed, 468 insertions(+), 45 deletions(-) create mode 100644 tests/test_live_services.py create mode 100644 tests/test_runtime_trigger_workflow.py diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index dce91e40..1e6a0a19 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -1,10 +1,6 @@ name: Runtime on: - workflow_run: - workflows: [CI] - types: [completed] - branches: [main] workflow_dispatch: inputs: validate_only: @@ -27,13 +23,6 @@ concurrency: jobs: deploy: - if: > - github.event_name == 'workflow_dispatch' || - ( - github.event_name == 'workflow_run' && - github.event.workflow_run.conclusion == 'success' && - github.event.workflow_run.event == 'push' - ) runs-on: self-hosted timeout-minutes: 60 environment: binance-runtime @@ -67,8 +56,6 @@ jobs: fi - name: 1. Checkout latest code uses: actions/checkout@v6 - with: - ref: ${{ github.event_name == 'workflow_run' && github.event.workflow_run.head_sha || github.sha }} - name: 2. Authenticate to Google Cloud uses: google-github-actions/auth@v3 @@ -171,6 +158,7 @@ jobs: BINANCE_API_KEY: ${{ secrets.BINANCE_API_KEY }} BINANCE_API_SECRET: ${{ secrets.BINANCE_API_SECRET }} STRATEGY_PROFILE: ${{ vars.STRATEGY_PROFILE || 'crypto_live_pool_rotation' }} + RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON }} TG_TOKEN: ${{ secrets.TG_TOKEN }} GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} NOTIFY_LANG: ${{ vars.NOTIFY_LANG }} @@ -271,9 +259,20 @@ jobs: "sha: ${GITHUB_SHA}" \ "actor: ${GITHUB_ACTOR}" \ "url: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}") - curl -fsS -X POST "https://api.telegram.org/bot${TG_TOKEN}/sendMessage" \ + RESPONSE="$(curl -fsS -X POST "https://api.telegram.org/bot${TG_TOKEN}/sendMessage" \ --data-urlencode "chat_id=${GLOBAL_TELEGRAM_CHAT_ID}" \ - --data-urlencode "text=${MESSAGE}" >/dev/null + --data-urlencode "text=${MESSAGE}")" + python3 - "${RESPONSE}" <<'PY' + import json + import sys + + try: + payload = json.loads(sys.argv[1]) + except json.JSONDecodeError as exc: + raise SystemExit("Telegram returned invalid JSON") from exc + if not isinstance(payload, dict) or payload.get("ok") is not True: + raise SystemExit("Telegram did not acknowledge workflow failure notification") + PY env: TG_TOKEN: ${{ secrets.TG_TOKEN }} GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} diff --git a/application/cycle_service.py b/application/cycle_service.py index f6397b2b..4f567137 100644 --- a/application/cycle_service.py +++ b/application/cycle_service.py @@ -8,6 +8,7 @@ from quant_platform_kit.common.runtime_reports import persist_runtime_report from quant_platform_kit.strategy_lifecycle.performance_monitor import try_record_platform_execution from runtime_logging import RuntimeLogContext, emit_runtime_log +from runtime_support import finalize_notification_delivery def execute_strategy_cycle( @@ -224,6 +225,7 @@ def execute_strategy_cycle( pass finally: report["log_lines"] = list(log_buffer) + finalize_notification_delivery(report) try_record_platform_execution( str(getattr(runtime, "strategy_profile", "") or ""), { @@ -269,10 +271,15 @@ def run_live_cycle( exit_fn=None, ): runtime = runtime_builder() + runtime_target = getattr(runtime, "runtime_target", None) log_context = RuntimeLogContext( platform="binance", deploy_target=os.getenv("LOG_DEPLOY_TARGET", "vps"), - service_name=os.getenv("SERVICE_NAME", "binance-platform"), + service_name=( + getattr(runtime_target, "service_name", None) + or os.getenv("SERVICE_NAME") + or "binance-platform" + ), strategy_profile=str(getattr(runtime, "strategy_profile", "") or os.getenv("STRATEGY_PROFILE", "crypto_live_pool_rotation")), run_id=str(getattr(runtime, "run_id", "") or ""), extra_fields={ diff --git a/docs/operator_runbook.md b/docs/operator_runbook.md index ba7ccfa2..4d5ca2fe 100644 --- a/docs/operator_runbook.md +++ b/docs/operator_runbook.md @@ -65,6 +65,7 @@ The monthly execution pool is locked to the accepted upstream `version` / `as_of - GitHub Actions no longer owns the hourly cadence for runtime execution in this repo. - Production cadence should come from one external scheduler, for example VPS cron calling the GitHub Actions dispatch API. - The VPS dispatch guard retries bounded transient failures such as network errors and GitHub `500`/`502`/`503`/`504`, but still alerts immediately for configuration and permission failures. +- Repository variable changes are consumed by the next externally scheduled dispatch; they do not reconfigure the VPS scheduler cadence. - Avoid overlapping dispatches from multiple schedulers or from a second manual run while the current runtime job is still in progress. ## Degraded Mode Ladder @@ -93,6 +94,7 @@ Use the generic `STRATEGY_ARTIFACT_*` names for crypto strategy artifacts. Primary settings: +- `RUNTIME_TARGET_JSON`: canonical runtime target written by QuantRuntimeSettings; when present, `STRATEGY_PROFILE` and `BINANCE_DRY_RUN` must match it or the run fails closed - `STRATEGY_PROFILE`: live profile selector; current supported value is `crypto_live_pool_rotation` - `STRATEGY_ARTIFACT_FIRESTORE_COLLECTION`: upstream artifact collection, default `strategy` - `STRATEGY_ARTIFACT_FIRESTORE_DOCUMENT`: upstream artifact document, default `CRYPTO_LIVE_POOL_ROTATION_LIVE_POOL` @@ -147,8 +149,10 @@ Operator action: Expected behavior: -- Telegram send failures should not stop the trading cycle -- The cycle may still finish while alert delivery is degraded +- Telegram transport and response-body acknowledgement are both validated +- A delivery failure does not roll back completed trading actions, but the execution report and workflow are marked failed +- Persisted notification receipts contain only delivery metadata and message hashes, never tokens, chat IDs, or message text +- A failed periodic status delivery does not advance its report bucket, so a later cycle can retry Operator action: diff --git a/live_services.py b/live_services.py index 39642301..5aeaeddb 100644 --- a/live_services.py +++ b/live_services.py @@ -1,3 +1,5 @@ +import hashlib + import requests from notify_i18n_support import build_telegram_message, translate as t @@ -46,10 +48,33 @@ def save_trade_state(data, *, normalize_fn, collection="strategy", document="MUL def send_tg_msg(token, chat_id, text): + message = build_telegram_message(text) + receipt = { + "sink": "telegram", + "delivery_status": "failed", + "transport_acknowledged": False, + "compact_text_sha256": hashlib.sha256(message.encode("utf-8")).hexdigest(), + "compact_text_length": len(message), + } if not token or not chat_id: - return + return {**receipt, "error_type": "missing_target"} url = f"https://api.telegram.org/bot{token}/sendMessage" try: - requests.post(url, data={"chat_id": chat_id, "text": build_telegram_message(text)}, timeout=10) - except Exception: + response = requests.post( + url, + data={"chat_id": chat_id, "text": message}, + timeout=10, + ) + if int(getattr(response, "status_code", 500)) >= 400: + return {**receipt, "error_type": "http_error"} + payload = response.json() + if not isinstance(payload, dict) or payload.get("ok") is not True: + return {**receipt, "error_type": "telegram_rejected"} + return { + **receipt, + "delivery_status": "sent", + "transport_acknowledged": True, + } + except Exception as exc: print(t("telegram_send_failed")) + return {**receipt, "error_type": type(exc).__name__} diff --git a/main.py b/main.py index 7061cc79..e3db73d1 100644 --- a/main.py +++ b/main.py @@ -430,7 +430,7 @@ def append_log(log_buffer, message): def send_tg_msg(token, chat_id, text): - live_send_tg_msg(token, chat_id, text) + return live_send_tg_msg(token, chat_id, text) def _runtime_error_notification_message(exc): @@ -454,11 +454,13 @@ def _notify_runtime_error(exc): print("Binance runtime error notification skipped: no Telegram target configured.") return False try: - send_tg_msg(token, chat_id, _runtime_error_notification_message(exc)) + receipt = send_tg_msg(token, chat_id, _runtime_error_notification_message(exc)) except Exception as send_exc: print(f"Binance runtime error Telegram send failed: {send_exc}") return False - return True + if isinstance(receipt, dict): + return receipt.get("transport_acknowledged") is True + return receipt is True # ========================================== # 2. Earn and balance helpers diff --git a/reporting/status_reports.py b/reporting/status_reports.py index 74ca3fee..d1b7978f 100644 --- a/reporting/status_reports.py +++ b/reporting/status_reports.py @@ -67,11 +67,15 @@ def maybe_send_periodic_btc_status_report( f"{separator}\n" f"💡 {hint}" ) - if notifier_fn is None: - send_tg_msg_fn(tg_token, tg_chat_id, text) + delivery = send_tg_msg_fn(tg_token, tg_chat_id, text) if notifier_fn is None else notifier_fn(text) + if isinstance(delivery, dict): + acknowledged = delivery.get("transport_acknowledged") is True else: - notifier_fn(text) + acknowledged = delivery is not False + if not acknowledged: + return False state["last_btc_status_report_bucket"] = report_bucket + return True def append_portfolio_report( diff --git a/runtime_config_support.py b/runtime_config_support.py index dca2552f..39d56186 100644 --- a/runtime_config_support.py +++ b/runtime_config_support.py @@ -5,6 +5,12 @@ from datetime import datetime, timezone from typing import Any, Callable +from quant_platform_kit.common.runtime_target import ( + RuntimeTarget, + build_runtime_target, + resolve_runtime_target_from_env, +) + from notify_i18n_support import build_strategy_display_name, build_translator, get_notify_lang from runtime_support import ExecutionRuntime from strategy_registry import ( @@ -43,14 +49,12 @@ class CycleExecutionSettings: strategy_display_name: str strategy_display_name_localized: str strategy_domain: str + runtime_target: RuntimeTarget def load_cycle_execution_settings() -> CycleExecutionSettings: notify_lang = get_notify_lang() - strategy_definition = resolve_strategy_definition( - os.getenv("STRATEGY_PROFILE"), - platform_id=BINANCE_PLATFORM, - ) + runtime_target, strategy_definition = _resolve_runtime_target() strategy_metadata = resolve_strategy_metadata( strategy_definition.profile, platform_id=BINANCE_PLATFORM, @@ -70,7 +74,53 @@ def load_cycle_execution_settings() -> CycleExecutionSettings: strategy_display_name=strategy_metadata.display_name, strategy_display_name_localized=strategy_display_name_localized, strategy_domain=strategy_definition.domain, + runtime_target=runtime_target, + ) + + +def _resolve_runtime_target(): + raw_runtime_target = str(os.getenv("RUNTIME_TARGET_JSON") or "").strip() + raw_strategy_profile = str(os.getenv("STRATEGY_PROFILE") or "").strip() + if raw_runtime_target: + runtime_target = resolve_runtime_target_from_env( + env=os.environ, + expected_platform_id=BINANCE_PLATFORM, + ) + strategy_definition = resolve_strategy_definition( + runtime_target.strategy_profile, + platform_id=BINANCE_PLATFORM, + ) + if raw_strategy_profile: + legacy_definition = resolve_strategy_definition( + raw_strategy_profile, + platform_id=BINANCE_PLATFORM, + ) + if legacy_definition.profile != strategy_definition.profile: + raise ValueError( + "STRATEGY_PROFILE does not match RUNTIME_TARGET_JSON.strategy_profile" + ) + if "BINANCE_DRY_RUN" in os.environ: + legacy_dry_run = get_env_bool("BINANCE_DRY_RUN", default=True) + if legacy_dry_run != runtime_target.dry_run_only: + raise ValueError( + "BINANCE_DRY_RUN does not match RUNTIME_TARGET_JSON.dry_run_only" + ) + return runtime_target, strategy_definition + + strategy_definition = resolve_strategy_definition( + raw_strategy_profile or None, + platform_id=BINANCE_PLATFORM, + ) + runtime_target = build_runtime_target( + platform_id=BINANCE_PLATFORM, + strategy_profile=strategy_definition.profile, + dry_run_only=get_env_bool("BINANCE_DRY_RUN", default=True), + deployment_selector=os.getenv("DEPLOYMENT_SELECTOR") or "default", + account_selector=os.getenv("ACCOUNT_SELECTOR") or "default", + account_scope=os.getenv("ACCOUNT_SCOPE") or "default", + service_name=os.getenv("SERVICE_NAME") or "binance-platform", ) + return runtime_target, strategy_definition def build_live_runtime( @@ -83,7 +133,7 @@ def build_live_runtime( runtime_now = now_utc or datetime.now(timezone.utc) cycle_settings = load_cycle_execution_settings() return ExecutionRuntime( - dry_run=get_env_bool("BINANCE_DRY_RUN", default=True), + dry_run=cycle_settings.runtime_target.dry_run_only, now_utc=runtime_now, strategy_profile=cycle_settings.strategy_profile, strategy_domain=cycle_settings.strategy_domain, @@ -96,4 +146,5 @@ def build_live_runtime( state_loader=state_loader, state_writer=state_writer, notifier=notifier, + runtime_target=cycle_settings.runtime_target, ) diff --git a/runtime_support.py b/runtime_support.py index b7291eb4..08450977 100644 --- a/runtime_support.py +++ b/runtime_support.py @@ -1,5 +1,7 @@ +import hashlib import os import time +from collections.abc import Mapping from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Callable, Optional @@ -37,6 +39,7 @@ class ExecutionRuntime: state_loader: Optional[Callable[..., Any]] = None state_writer: Optional[Callable[[dict[str, Any]], Any]] = None notifier: Optional[Callable[..., Any]] = None + runtime_target: Any = None trend_pool_payload: Optional[dict[str, Any]] = None btc_market_snapshot: Optional[dict[str, Any]] = None trend_indicator_snapshots: Optional[dict[str, Any]] = None @@ -52,10 +55,16 @@ def __post_init__(self): def build_execution_report(runtime): + runtime_target = getattr(runtime, "runtime_target", None) + runtime_service_name = ( + getattr(runtime_target, "service_name", None) + or os.getenv("SERVICE_NAME") + or "binance-platform" + ) report = build_runtime_report_base( platform="binance", deploy_target=os.getenv("LOG_DEPLOY_TARGET", "vps"), - service_name=os.getenv("SERVICE_NAME", "binance-platform"), + service_name=runtime_service_name, strategy_profile=str(runtime.strategy_profile or os.getenv("STRATEGY_PROFILE", "crypto_live_pool_rotation")), strategy_domain=str(runtime.strategy_domain or os.getenv("STRATEGY_DOMAIN", "crypto")), run_id=str(runtime.run_id), @@ -97,6 +106,8 @@ def build_execution_report(runtime): "strategy_display_name_localized": str(runtime.strategy_display_name_localized or ""), }, }) + if runtime_target is not None: + report["runtime_target"] = runtime_target.to_dict() return report @@ -141,21 +152,99 @@ def next_order_id(runtime, prefix, symbol): def runtime_notify(runtime, report, text): - payload = { - "token": str(runtime.tg_token), - "chat_id": str(runtime.tg_chat_id), - "text": str(text), + message = str(text) + safe_event = { + "sink": "telegram", + "compact_text_sha256": hashlib.sha256(message.encode("utf-8")).hexdigest(), + "compact_text_length": len(message), "run_id": str(runtime.run_id), "dry_run": bool(runtime.dry_run), } - report["notifications"].append(payload) if runtime.dry_run: - record_side_effect(runtime, report, effect_type="notify", target="telegram", payload=payload, executed=False) - return + safe_event.update( + { + "delivery_status": "suppressed", + "transport_acknowledged": False, + } + ) + report["notifications"].append(safe_event) + record_side_effect( + runtime, + report, + effect_type="notify", + target="telegram", + payload=safe_event, + executed=False, + ) + return False if runtime.notifier is None: raise RuntimeError("runtime.notifier is not configured") - runtime.notifier(**payload) - record_side_effect(runtime, report, effect_type="notify", target="telegram", payload=payload, executed=True) + receipt = runtime.notifier( + token=str(runtime.tg_token), + chat_id=str(runtime.tg_chat_id), + text=message, + run_id=str(runtime.run_id), + dry_run=False, + ) + if isinstance(receipt, Mapping): + for key in ( + "sink", + "delivery_status", + "transport_acknowledged", + "error_type", + "compact_text_sha256", + "compact_text_length", + ): + if key in receipt: + safe_event[key] = receipt[key] + acknowledged = receipt.get("transport_acknowledged") is True + else: + acknowledged = receipt is True + safe_event.setdefault("delivery_status", "sent" if acknowledged else "failed") + safe_event["transport_acknowledged"] = acknowledged + report["notifications"].append(safe_event) + delivery_events = [ + event + for event in report["notifications"] + if event.get("delivery_status") != "suppressed" + ] + report.setdefault("summary", {})["notification_delivery_summary"] = { + "event_count": len(delivery_events), + "sent_count": sum( + event.get("transport_acknowledged") is True for event in delivery_events + ), + "failed_count": sum( + event.get("transport_acknowledged") is not True for event in delivery_events + ), + "all_acknowledged": all( + event.get("transport_acknowledged") is True for event in delivery_events + ), + } + record_side_effect( + runtime, + report, + effect_type="notify", + target="telegram", + payload=safe_event, + executed=acknowledged, + ) + return acknowledged + + +def finalize_notification_delivery(report): + delivery_summary = report.get("summary", {}).get("notification_delivery_summary") + if not isinstance(delivery_summary, dict) or delivery_summary.get("all_acknowledged") is not False: + return + errors = report.setdefault("error_summary", {}).setdefault("errors", []) + if not any(error.get("stage") == "notification_delivery" for error in errors if isinstance(error, dict)): + errors.append( + { + "stage": "notification_delivery", + "message": "Telegram delivery was not acknowledged.", + } + ) + if report.get("status") == "ok": + report["status"] = "error" def runtime_set_trade_state(runtime, report, state, *, reason): diff --git a/scripts/runtime_workflow_heartbeat.py b/scripts/runtime_workflow_heartbeat.py index 9f9df99f..dd16eab6 100644 --- a/scripts/runtime_workflow_heartbeat.py +++ b/scripts/runtime_workflow_heartbeat.py @@ -206,7 +206,16 @@ def _send_telegram(message: str) -> bool: ) try: with urllib.request.urlopen(request, timeout=15) as response: - ok = ok and response.status < 400 + try: + payload = json.loads(response.read().decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + payload = {} + ok = ( + ok + and response.status < 400 + and isinstance(payload, dict) + and payload.get("ok") is True + ) except Exception as exc: # noqa: BLE001 ok = False print(f"Telegram send failed: {exc}", file=sys.stderr) diff --git a/tests/test_live_services.py b/tests/test_live_services.py new file mode 100644 index 00000000..5d23e097 --- /dev/null +++ b/tests/test_live_services.py @@ -0,0 +1,31 @@ +from unittest.mock import Mock, patch + +from live_services import send_tg_msg + + +def test_send_tg_msg_rejects_telegram_ok_false(): + response = Mock(status_code=200) + response.json.return_value = {"ok": False, "description": "rejected"} + + with patch("live_services.requests.post", return_value=response): + receipt = send_tg_msg("token-value", "chat-value", "hello") + + assert receipt["delivery_status"] == "failed" + assert receipt["transport_acknowledged"] is False + assert receipt["error_type"] == "telegram_rejected" + assert "token-value" not in str(receipt) + assert "chat-value" not in str(receipt) + assert "hello" not in str(receipt) + + +def test_send_tg_msg_records_safe_acknowledged_receipt(): + response = Mock(status_code=200) + response.json.return_value = {"ok": True, "result": {"message_id": 123}} + + with patch("live_services.requests.post", return_value=response): + receipt = send_tg_msg("token-value", "chat-value", "hello") + + assert receipt["delivery_status"] == "sent" + assert receipt["transport_acknowledged"] is True + assert len(receipt["compact_text_sha256"]) == 64 + assert receipt["compact_text_length"] > 0 diff --git a/tests/test_main_runtime_error_notification.py b/tests/test_main_runtime_error_notification.py index abb3b9b6..3b05b306 100644 --- a/tests/test_main_runtime_error_notification.py +++ b/tests/test_main_runtime_error_notification.py @@ -144,6 +144,25 @@ def fake_send_tg_msg(token, chat_id, text): self.assertIn("Binance strategy run failed", observed["messages"][0][2]) self.assertIn("RuntimeError: runtime setup failed", observed["messages"][0][2]) + def test_runtime_error_notification_requires_transport_acknowledgement(self): + with patch.dict( + os.environ, + { + "TG_TOKEN": "token-1", + "GLOBAL_TELEGRAM_CHAT_ID": "chat-1", + }, + clear=False, + ): + with patch.object( + main, + "send_tg_msg", + return_value={ + "delivery_status": "failed", + "transport_acknowledged": False, + }, + ): + self.assertFalse(main._notify_runtime_error(RuntimeError("boom"))) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index e54c39bf..4213b8e5 100644 --- a/tests/test_runtime_config_support.py +++ b/tests/test_runtime_config_support.py @@ -147,6 +147,54 @@ def test_build_live_runtime_uses_global_telegram_chat_id(self): self.assertEqual(runtime.tg_chat_id, "shared-chat-id") + def test_build_live_runtime_uses_runtime_target_json(self): + runtime_target = { + "platform_id": "binance", + "strategy_profile": DEFAULT_STRATEGY_PROFILE, + "dry_run_only": False, + "deployment_selector": "default", + "account_selector": ["default"], + "account_scope": "default", + "service_name": "binance-platform", + "execution_mode": "live", + "market": "CRYPTO", + "market_calendar": "24/7", + "market_timezone": "UTC", + } + with patch.dict( + os.environ, + { + "RUNTIME_TARGET_JSON": json.dumps(runtime_target), + "STRATEGY_PROFILE": DEFAULT_STRATEGY_PROFILE, + "BINANCE_DRY_RUN": "false", + }, + clear=True, + ): + runtime = build_live_runtime() + + self.assertFalse(runtime.dry_run) + self.assertEqual(runtime.runtime_target.service_name, "binance-platform") + self.assertEqual(runtime.runtime_target.strategy_profile, DEFAULT_STRATEGY_PROFILE) + + def test_runtime_target_and_legacy_dry_run_variable_must_match(self): + runtime_target = { + "platform_id": "binance", + "strategy_profile": DEFAULT_STRATEGY_PROFILE, + "dry_run_only": False, + "execution_mode": "live", + } + with patch.dict( + os.environ, + { + "RUNTIME_TARGET_JSON": json.dumps(runtime_target), + "STRATEGY_PROFILE": DEFAULT_STRATEGY_PROFILE, + "BINANCE_DRY_RUN": "true", + }, + clear=True, + ): + with self.assertRaisesRegex(ValueError, "BINANCE_DRY_RUN"): + build_live_runtime() + def test_status_script_json_matches_registry(self): script = ROOT / "scripts" / "print_strategy_profile_status.py" result = subprocess.run( diff --git a/tests/test_runtime_support.py b/tests/test_runtime_support.py index 248e8ac0..ef6deb0e 100644 --- a/tests/test_runtime_support.py +++ b/tests/test_runtime_support.py @@ -12,7 +12,14 @@ if str(QPK_SRC) not in sys.path: sys.path.insert(0, str(QPK_SRC)) -from runtime_support import ExecutionRuntime, build_execution_report, record_gating_event +from runtime_support import ( + ExecutionRuntime, + build_execution_report, + finalize_notification_delivery, + record_gating_event, + runtime_notify, +) +from quant_platform_kit.common.runtime_target import build_runtime_target class TestBuildExecutionReport(unittest.TestCase): @@ -48,6 +55,24 @@ def test_report_preserves_existing_fields(self): self.assertIn("buy_sell_intents", report) self.assertIn("log_lines", report) + def test_report_uses_runtime_target_service_identity(self): + runtime_target = build_runtime_target( + platform_id="binance", + strategy_profile="crypto_live_pool_rotation", + dry_run_only=False, + service_name="binance-platform", + ) + runtime = ExecutionRuntime( + dry_run=False, + run_id="test-runtime-target", + runtime_target=runtime_target, + ) + + report = build_execution_report(runtime) + + self.assertEqual(report["service_name"], "binance-platform") + self.assertEqual(report["runtime_target"]["service_name"], "binance-platform") + def test_record_gating_event_updates_summary_and_events(self): report = {} @@ -67,3 +92,31 @@ def test_record_gating_event_updates_summary_and_events(self): self.assertEqual(report["gating_summary"]["trend_buy_below_min_budget"], 2) self.assertEqual(report["gating_events"][0]["symbol"], "ETHUSDT") self.assertEqual(report["gating_events"][0]["detail"]["budget_usdt"], 12.0) + + def test_runtime_notify_persists_only_safe_failed_delivery_receipt(self): + runtime = ExecutionRuntime( + dry_run=False, + run_id="test-notification", + tg_token="secret-token", + tg_chat_id="private-chat", + notifier=lambda **_kwargs: { + "sink": "telegram", + "delivery_status": "failed", + "transport_acknowledged": False, + "error_type": "telegram_rejected", + }, + ) + report = build_execution_report(runtime) + + acknowledged = runtime_notify(runtime, report, "sensitive notification body") + finalize_notification_delivery(report) + + serialized = str(report) + self.assertFalse(acknowledged) + self.assertEqual(report["status"], "error") + self.assertFalse( + report["summary"]["notification_delivery_summary"]["all_acknowledged"] + ) + self.assertNotIn("secret-token", serialized) + self.assertNotIn("private-chat", serialized) + self.assertNotIn("sensitive notification body", serialized) diff --git a/tests/test_runtime_trigger_workflow.py b/tests/test_runtime_trigger_workflow.py new file mode 100644 index 00000000..ded608b3 --- /dev/null +++ b/tests/test_runtime_trigger_workflow.py @@ -0,0 +1,12 @@ +from pathlib import Path + + +WORKFLOW = Path(__file__).resolve().parents[1] / ".github" / "workflows" / "main.yml" + + +def test_runtime_workflow_is_dispatch_only() -> None: + workflow = WORKFLOW.read_text(encoding="utf-8") + + assert "workflow_dispatch:" in workflow + assert "workflow_run:" not in workflow + assert "github.event.workflow_run" not in workflow diff --git a/tests/test_runtime_workflow_heartbeat.py b/tests/test_runtime_workflow_heartbeat.py index 98dc2937..20029a48 100644 --- a/tests/test_runtime_workflow_heartbeat.py +++ b/tests/test_runtime_workflow_heartbeat.py @@ -121,6 +121,27 @@ def fake_github_request(url: str, token: str) -> dict[str, object]: self.assertTrue(any("/actions/runs?" in url for url in requested_urls)) send_telegram.assert_not_called() + def test_send_telegram_rejects_api_ok_false(self) -> None: + class FakeResponse: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self): + return b'{"ok":false,"description":"rejected"}' + + with patch.dict( + os.environ, + {"TG_TOKEN": "token-1", "GLOBAL_TELEGRAM_CHAT_ID": "chat-1"}, + clear=True, + ): + with patch.object(heartbeat.urllib.request, "urlopen", return_value=FakeResponse()): + self.assertFalse(heartbeat._send_telegram("heartbeat failed")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_runtime_workflow_shared_config.sh b/tests/test_runtime_workflow_shared_config.sh index ac44f56f..ea5b762c 100644 --- a/tests/test_runtime_workflow_shared_config.sh +++ b/tests/test_runtime_workflow_shared_config.sh @@ -8,6 +8,7 @@ grep -Fq 'TG_TOKEN: ${{ secrets.TG_TOKEN }}' "$workflow_file" grep -Fq 'GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }}' "$workflow_file" grep -Fq 'NOTIFY_LANG: ${{ vars.NOTIFY_LANG }}' "$workflow_file" grep -Fq "STRATEGY_PROFILE: \${{ vars.STRATEGY_PROFILE || 'crypto_live_pool_rotation' }}" "$workflow_file" +grep -Fq 'RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON }}' "$workflow_file" grep -Fq 'id-token: write' "$workflow_file" grep -Fq 'workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}' "$workflow_file" grep -Fq 'service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}' "$workflow_file" @@ -27,6 +28,7 @@ grep -Fq '7. Notify Telegram on runtime workflow failure' "$workflow_file" grep -Fq "if: \${{ failure() && github.event.inputs.validate_only != 'true' }}" "$workflow_file" grep -Fq 'reports/execution_report.json' "$workflow_file" grep -Fq 'https://api.telegram.org/bot${TG_TOKEN}/sendMessage' "$workflow_file" +grep -Fq 'payload.get("ok") is not True' "$workflow_file" if grep -Fq 'TG_CHAT_ID:' "$workflow_file"; then echo "workflow should not pass TG_CHAT_ID anymore" >&2 exit 1 diff --git a/tests/test_status_reports.py b/tests/test_status_reports.py index 6636572c..b7c26372 100644 --- a/tests/test_status_reports.py +++ b/tests/test_status_reports.py @@ -125,6 +125,50 @@ def translate_strategy(key, **kwargs): self.assertTrue(observed) self.assertIn("strategy=加密领涨轮动", observed[0]) + def test_periodic_report_does_not_advance_bucket_when_delivery_fails(self): + state = {} + now = SimpleNamespace(hour=8, strftime=lambda _fmt: "2026032908") + translations = { + "heartbeat_title": "heartbeat", + "strategy_label": "strategy={name}", + "total_equity": "equity", + "trend_equity": "trend", + "btc_price": "btc", + "btc_gate": "gate", + "gate_on": "on", + "gate_off": "off", + "zscore": "zscore", + "zscore_threshold": "threshold", + "btc_target": "target", + "manual_hint_low_value": "low", + "manual_hint_neutral": "neutral", + } + + def translate(key, **kwargs): + template = translations[key] + return template.format(**kwargs) if kwargs else template + + acknowledged = maybe_send_periodic_btc_status_report( + state, + "token", + "chat-id", + now, + 8, + 1000.0, + 250.0, + 0.01, + 50_000.0, + {"ahr999": 0.8, "zscore": 1.0, "sell_trigger": 3.0, "regime_on": True}, + 0.5, + "Crypto", + translate_fn=translate, + separator="---", + notifier_fn=lambda _text: False, + ) + + self.assertFalse(acknowledged) + self.assertNotIn("last_btc_status_report_bucket", state) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py index d93e634c..f305b846 100644 --- a/tests/test_strategy_runtime.py +++ b/tests/test_strategy_runtime.py @@ -36,7 +36,10 @@ def test_load_strategy_runtime_exposes_explicit_artifact_contract(self): self.assertEqual(runtime.profile, "crypto_live_pool_rotation") self.assertEqual(runtime.runtime_adapter.portfolio_input_name, "portfolio_snapshot") - self.assertTrue(str(runtime.default_local_artifact_path).endswith("BinancePlatform/artifacts/live_pool_legacy.json")) + self.assertEqual( + runtime.default_local_artifact_path, + PROJECT_ROOT / "artifacts" / "live_pool_legacy.json", + ) self.assertEqual(runtime.artifact_contract["version"], "crypto_live_pool_rotation.live_pool.v1") self.assertTrue(runtime.artifact_contract["requires_artifacts"]) self.assertTrue(runtime.artifact_contract["requires_manifest"]) From eb36e0f7cc5f1087e0dc759fb43bc9032225ec7f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:48:25 +0800 Subject: [PATCH 2/2] test: enforce dispatch-only runtime trigger Co-Authored-By: Codex --- tests/test_runtime_trigger_workflow.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/test_runtime_trigger_workflow.py b/tests/test_runtime_trigger_workflow.py index ded608b3..04cd9f9e 100644 --- a/tests/test_runtime_trigger_workflow.py +++ b/tests/test_runtime_trigger_workflow.py @@ -1,12 +1,14 @@ from pathlib import Path +import unittest WORKFLOW = Path(__file__).resolve().parents[1] / ".github" / "workflows" / "main.yml" -def test_runtime_workflow_is_dispatch_only() -> None: - workflow = WORKFLOW.read_text(encoding="utf-8") +class RuntimeTriggerWorkflowTest(unittest.TestCase): + def test_runtime_workflow_is_dispatch_only(self) -> None: + workflow = WORKFLOW.read_text(encoding="utf-8") - assert "workflow_dispatch:" in workflow - assert "workflow_run:" not in workflow - assert "github.event.workflow_run" not in workflow + self.assertIn("workflow_dispatch:", workflow) + self.assertNotIn("workflow_run:", workflow) + self.assertNotIn("github.event.workflow_run", workflow)