main.py

"""
LongPort Cloud Run strategy executor.
Runs the configured shared UsEquityStrategies profile for the deployed LongBridge account.
Runs on Cloud Run; token from Secret Manager, orders via LongPort OpenAPI, alerts via Telegram.
"""
import dataclasses
import json
import importlib
import os
import time
from datetime import datetime, timezone

from flask import Flask, request

import google.auth
import requests
from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets
from application.startup_config import (
    StartupConfigReceipt,
    normalize_monitor_dispatch_targets,
    normalize_plugin_mounts_strategy,
    overlay_env,
)
from application.broker_reconciliation import (
    account_snapshot_enabled,
    collect_read_only_reconciliation_observations,
    reconciliation_enabled,
    run_read_only_account_snapshot,
    run_read_only_broker_reconciliation,
)
from application.execution_receipt_adapter import (
    attach_cycle_execution_receipt,
    attach_terminal_fallback_execution_receipt,
)
from application.runtime_broker_adapters import build_runtime_broker_adapters
from application.runtime_composer import build_runtime_composer
from application.rebalance_service import run_strategy as run_rebalance_cycle
from application.durable_execution_commands import (
    resolve_paper_execution_command_consumer_enabled,
)
from application.paper_execution_command_consumer import (
    consume_due_paper_execution_commands,
)
from application.runtime_strategy_adapters import build_runtime_strategy_adapters
from application.longbridge_execution import submit_order
from runtime_execution_policy import fractional_buy_execution_enabled, FRACTIONAL_BUY_QUANTITY_STEP
from application.longbridge_portfolio import fetch_strategy_account_state
from application.account_snapshot import (
    begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history,
    read_cycle_account_balance,
)
from scripts.record_daily_account_snapshot import record_projected_daily_account
from entrypoints.cloud_run import is_market_open_now
from runtime_execution_policy import dca_execution_unsupported_reason
from runtime_config_support import load_platform_runtime_settings
from notifications.telegram import build_signal_text, build_strategy_display_name, build_translator
from notifications import unified as unified_notifications
from quant_platform_kit.common.runtime_reports import (
    append_runtime_report_error,
    build_runtime_report_base,
    finalize_runtime_report,
    persist_runtime_report,
)
from quant_platform_kit.common.strategy_plugins import (
    build_strategy_plugin_report_payload,
    load_configured_strategy_plugin_signals,
    parse_strategy_plugin_mounts,
)
from quant_platform_kit.notifications.strategy_plugin_alerts import (
    StrategyPluginAlertStateSettings,
    build_strategy_plugin_alert_context_label as build_alert_context_label,
    publish_strategy_plugin_alerts as dispatch_strategy_plugin_alerts,
)
from quant_platform_kit.common.runtime_inputs import build_strategy_evaluation_inputs
from runtime_logging import build_run_id, emit_runtime_log
from quant_platform_kit.longbridge import (
    build_contexts,
    calculate_rotation_indicators,
    estimate_max_purchase_quantity,
    fetch_last_price,
    fetch_order_status,
    fetch_token_from_secret,
    refresh_token_if_needed,
)
from strategy_runtime import load_strategy_runtime
from decision_mapper import map_strategy_decision_to_plan

app = Flask(__name__)

# This collector is limited to documented LongPort read endpoints and is
# independently fail-closed for incomplete account identity/order coverage.
READ_ONLY_BROKER_RECONCILIATION_COLLECTOR = collect_read_only_reconciliation_observations

# ---------------------------------------------------------------------------
# Config and constants (GCP project, Telegram, execution and strategy params)
# ---------------------------------------------------------------------------
def get_project_id():
    try:
        _, project_id = google.auth.default()
        return project_id
    except Exception:
        return os.getenv("GOOGLE_CLOUD_PROJECT")

RUNTIME_SETTINGS = load_platform_runtime_settings(project_id_resolver=get_project_id)
PROJECT_ID = RUNTIME_SETTINGS.project_id
SECRET_NAME = RUNTIME_SETTINGS.secret_name
ACCOUNT_PREFIX = RUNTIME_SETTINGS.account_prefix
STRATEGY_PROFILE = RUNTIME_SETTINGS.strategy_profile
STRATEGY_DISPLAY_NAME = RUNTIME_SETTINGS.strategy_display_name


def _normalize_plugin_mounts_strategy(raw_mounts_json: str | None) -> str | None:
    normalized, _corrections = normalize_plugin_mounts_strategy(
        raw_mounts_json,
        strategy_profile=STRATEGY_PROFILE,
    )
    return normalized


def _adopt_startup_config(settings, env):
    """Explicitly normalize startup config without mutating ``os.environ`` (B08).

    Returns ``(settings, monitor_dispatch_env_overrides, receipt)``.
    """
    corrections = []
    if hasattr(settings, "strategy_plugin_mounts_json"):
        normalized, plugin_corrections = normalize_plugin_mounts_strategy(
            settings.strategy_plugin_mounts_json,
            strategy_profile=STRATEGY_PROFILE,
        )
        corrections.extend(plugin_corrections)
        if normalized != settings.strategy_plugin_mounts_json:
            if dataclasses.is_dataclass(settings) and not isinstance(settings, type):
                settings = dataclasses.replace(settings, strategy_plugin_mounts_json=normalized)
            else:  # non-dataclass settings objects (test doubles)
                setattr(settings, "strategy_plugin_mounts_json", normalized)
    overrides, monitor_corrections = normalize_monitor_dispatch_targets(
        env,
        strategy_profile=STRATEGY_PROFILE,
    )
    corrections.extend(monitor_corrections)
    receipt = StartupConfigReceipt(
        strategy_profile=STRATEGY_PROFILE,
        corrections=tuple(corrections),
        env_overrides=tuple(sorted(overrides)),
    )
    return settings, overrides, receipt


RUNTIME_SETTINGS, MONITOR_DISPATCH_ENV_OVERRIDES, STARTUP_CONFIG_RECEIPT = _adopt_startup_config(
    RUNTIME_SETTINGS,
    os.environ,
)


def _load_monitor_dispatch_targets():
    # No corrections: identical call to before. Corrections: read through an
    # explicit overlay instead of the previously mutated ``os.environ``.
    if not MONITOR_DISPATCH_ENV_OVERRIDES:
        return load_monitor_targets()
    return load_monitor_targets(overlay_env(os.environ, MONITOR_DISPATCH_ENV_OVERRIDES))


ACCOUNT_REGION = RUNTIME_SETTINGS.account_region
MARKET = RUNTIME_SETTINGS.market
MARKET_CALENDAR = RUNTIME_SETTINGS.market_calendar
MARKET_TIMEZONE = RUNTIME_SETTINGS.market_timezone
SYMBOL_SUFFIX = RUNTIME_SETTINGS.symbol_suffix
TRADING_CURRENCY = RUNTIME_SETTINGS.trading_currency
NOTIFY_LANG = RUNTIME_SETTINGS.notify_lang
TG_TOKEN = RUNTIME_SETTINGS.tg_token
TG_CHAT_ID = RUNTIME_SETTINGS.tg_chat_id
from quant_platform_kit.notifications.cycle_channel import resolve_cycle_channel_and_url

_NOTIFICATION_CHANNEL, _NOTIFICATION_WEBHOOK_URL = resolve_cycle_channel_and_url(
    explicit_channel=RUNTIME_SETTINGS.notification_channel,
    telegram_token=TG_TOKEN,
    telegram_chat_id=TG_CHAT_ID,
    wecom_url=RUNTIME_SETTINGS.wecom_webhook_url,
    dingtalk_url=RUNTIME_SETTINGS.dingtalk_webhook_url,
    feishu_url=RUNTIME_SETTINGS.feishu_webhook_url,
    serverchan_url=RUNTIME_SETTINGS.serverchan_webhook_url,
)
CASH_ONLY_EXECUTION = getattr(RUNTIME_SETTINGS, "cash_only_execution", True)
STRATEGY_RUNTIME = load_strategy_runtime(
    STRATEGY_PROFILE,
    runtime_settings=RUNTIME_SETTINGS,
    logger=lambda message: print(message, flush=True),
)
STRATEGY_RUNTIME_CONFIG = dict(STRATEGY_RUNTIME.merged_runtime_config)
MANAGED_SYMBOLS = STRATEGY_RUNTIME.managed_symbols
AVAILABLE_INPUTS = frozenset(STRATEGY_RUNTIME.runtime_adapter.available_inputs)
BENCHMARK_SYMBOL = str(STRATEGY_RUNTIME_CONFIG.get("benchmark_symbol", "QQQ"))


def _emit_v7_loaded_version_log() -> None:
    """Emit one safe startup receipt for the disabled candidate revision."""

    if STRATEGY_PROFILE != "soxl_soxx_core_only_p2_v7_longterm_compounding_cash_reserve":
        return
    from application.v7_paper_application import (
        V7PaperApplicationError,
        build_v7_loaded_version_readback,
        load_v7_paper_application_binding,
    )

    binding = load_v7_paper_application_binding(os.environ)
    if binding is None:
        return
    try:
        readback = build_v7_loaded_version_readback(
            binding,
            runtime_settings=RUNTIME_SETTINGS,
            strategy_runtime=STRATEGY_RUNTIME,
            runtime_env=os.environ,
        )
        print(
            json.dumps(
                {"event": "v7_paper_application_loaded", "status": "applied_paused", "readback": readback},
                ensure_ascii=True,
                separators=(",", ":"),
            ),
            flush=True,
        )
    except V7PaperApplicationError:
        print(
            json.dumps(
                {"event": "v7_paper_application_loaded", "status": "uncertain"},
                ensure_ascii=True,
                separators=(",", ":"),
            ),
            flush=True,
        )


_emit_v7_loaded_version_log()
SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS = getattr(
    getattr(STRATEGY_RUNTIME.runtime_adapter, "runtime_policy", None),
    "signal_effective_after_trading_days",
    None,
)

# Order pricing: limit order discount/premium relative to last price
LIMIT_SELL_DISCOUNT = 0.995               # sell limit at 0.5% below last
LIMIT_BUY_PREMIUM = 1.005                 # buy limit at 0.5% above last
DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL = {"SOXL": 1.015, "TQQQ": 1.010}

# Order monitoring: poll interval and max attempts for fill check
ORDER_POLL_INTERVAL_SEC = 1
ORDER_POLL_MAX_ATTEMPTS = 8

# Token refresh: days before expiry to trigger refresh
TOKEN_REFRESH_THRESHOLD_DAYS = 30
DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD = 1000.0
DEFAULT_MIN_ORDER_NOTIONAL_USD = 100.0

_FRACTIONAL_BUY_EXECUTION = fractional_buy_execution_enabled(STRATEGY_PROFILE)


def _profile_submit_order(t_ctx, symbol, **kwargs):
    return submit_order(
        t_ctx,
        symbol,
        allow_fractional_shares=_FRACTIONAL_BUY_EXECUTION,
        quantity_step=FRACTIONAL_BUY_QUANTITY_STEP if _FRACTIONAL_BUY_EXECUTION else 1.0,
        **kwargs,
    )


def _profile_estimate_max_purchase_quantity(t_ctx, symbol, **kwargs):
    return estimate_max_purchase_quantity(
        t_ctx,
        symbol,
        fractional_shares=_FRACTIONAL_BUY_EXECUTION,
        **kwargs,
    )

SEPARATOR = "━━━━━━━━━━━━━━━━━━"
COMPACT_ERROR_NOTIFICATION_MAX_CHARS = 3500


def _load_limit_buy_premium_by_symbol(*env_names: str) -> dict[str, float]:
    raw_value = ""
    for env_name in env_names:
        value = os.getenv(env_name)
        if value and value.strip():
            raw_value = value.strip()
            break
    if not raw_value:
        return dict(DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL)
    try:
        payload = json.loads(raw_value)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Invalid limit buy premium map JSON: {raw_value!r}") from exc
    if not isinstance(payload, dict):
        raise ValueError("Limit buy premium map must be a JSON object keyed by symbol.")
    parsed: dict[str, float] = {}
    for symbol, premium in payload.items():
        symbol_text = str(symbol or "").strip().upper()
        if not symbol_text:
            continue
        premium_value = float(premium)
        if premium_value <= 0.0:
            raise ValueError(f"Limit buy premium for {symbol_text} must be positive.")
        parsed[symbol_text] = premium_value
    return parsed


LIMIT_BUY_PREMIUM_BY_SYMBOL = _load_limit_buy_premium_by_symbol(
    "LONGBRIDGE_LIMIT_BUY_PREMIUM_BY_SYMBOL_JSON",
    "LIMIT_BUY_PREMIUM_BY_SYMBOL_JSON",
)


def t(key, **kwargs):
    return build_translator(NOTIFY_LANG)(key, **kwargs)


def _runtime_failure_reason(exc: Exception) -> str:
    # Match only fixed platform errors; never publish arbitrary provider text.
    if isinstance(exc, RuntimeError) and str(exc) in {
        "durable execution command producer is paper-only and cannot be enabled live",
        "durable execution command consumer is paper-only and cannot be enabled live",
        "strategy risk state persistence is paper-only and cannot be enabled live",
    }:
        return "paper_live_conflict"
    return "unknown"


def _compact_error_notification(
    exc: Exception,
    *,
    title: str | None = None,
    prefix: str = "",
    phase: str = "strategy_cycle",
) -> str:
    error_text = type(exc).__name__
    known_phases = {"market_hours", "runtime_configuration", "strategy_cycle", "reporting", "health_probe"}
    safe_phase = phase if phase in known_phases else "strategy_cycle"
    reason = _runtime_failure_reason(exc)
    message = "\n".join((
        title or t("error_title"),
        t("failure_phase", phase=t(f"failure_phase_{safe_phase}")),
        t(f"failure_reason_{reason}"),
        f"{prefix}{error_text}",
    ))
    if len(message) <= COMPACT_ERROR_NOTIFICATION_MAX_CHARS:
        return message
    return message[: COMPACT_ERROR_NOTIFICATION_MAX_CHARS - 1].rstrip() + "…"


def _split_env_list(value: str | None) -> tuple[str, ...]:
    return tuple(
        item.strip()
        for item in str(value or "").replace(";", ",").split(",")
        if item.strip()
    )


def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict:
    if cycle_result is None:
        return {
            "action_done": False,
            "order_events_count": 0,
            "orders_previewed_count": 0,
            "orders_skipped_count": 0,
            "notes_count": 0,
            "dry_run_order_preview_available": False,
        }
    logs = tuple(getattr(cycle_result, "logs", ()) or ())
    skip_logs = tuple(getattr(cycle_result, "skip_logs", ()) or ())
    note_logs = tuple(getattr(cycle_result, "note_logs", ()) or ())
    dry_run_orders = tuple(getattr(cycle_result, "dry_run_orders", ()) or ())
    pending_orders = tuple(getattr(cycle_result, "pending_orders", ()) or ())
    quote_snapshots = tuple(getattr(cycle_result, "quote_snapshots", ()) or ())
    order_events_count = 0 if pending_orders else len(logs)
    orders_previewed_count = len(dry_run_orders) if dry_run_orders else (order_events_count if dry_run else 0)
    broker_submission_done = bool(getattr(cycle_result, "action_done", False))
    summary = {
        "action_done": broker_submission_done and not pending_orders,
        "broker_submission_done": broker_submission_done,
        "execution_status": (
            "pending_reconciliation"
            if pending_orders
            else ("previewed" if dry_run and broker_submission_done else "no_action")
        ),
        "order_events_count": order_events_count,
        "orders_pending_count": len(pending_orders),
        "orders_previewed_count": orders_previewed_count,
        "orders_skipped_count": len(skip_logs),
        "notes_count": len(note_logs),
        "dry_run_order_preview_available": bool(dry_run and orders_previewed_count > 0),
    }
    if dry_run_orders:
        summary["orders_previewed"] = [dict(order) for order in dry_run_orders]
    if pending_orders:
        summary["orders_pending"] = [dict(order) for order in pending_orders]
    if quote_snapshots:
        summary["quote_snapshot"] = {
            "quotes": [dict(snapshot) for snapshot in quote_snapshots],
        }
    execution = dict(getattr(cycle_result, "execution", {}) or {})
    account_snapshot = execution.get("heartbeat_account_snapshot")
    if isinstance(account_snapshot, dict):
        summary["heartbeat_account_snapshot"] = dict(account_snapshot)
    if execution.get("execution_status") == "blocked":
        summary["execution_status"] = "blocked"
        summary["blocked_reason"] = (
            "durable_live_execution_command_binding_invalid"
            if execution.get("blocked_reason") == "durable_live_execution_command_binding_invalid"
            else "unknown"
        )
    for field in (
        "signal_date",
        "effective_date",
        "execution_timing_contract",
        "execution_calendar_source",
    ):
        if execution.get(field) not in (None, ""):
            summary[field] = execution[field]
    durable_command = execution.get("durable_execution_command")
    if isinstance(durable_command, dict):
        summary["durable_execution_command"] = dict(durable_command)
    live_command = execution.get("durable_live_execution_command")
    if isinstance(live_command, dict):
        summary["durable_live_execution_command"] = dict(live_command)
    account_identity = execution.get("account_identity")
    if isinstance(account_identity, dict):
        summary["account_identity"] = dict(account_identity)
    from notifications.digest_snapshot.digest_routing import daily_digest_enabled
    if unified_notifications.enabled():
        _, summary["notification_digest"] = unified_notifications.cycle_facts(cycle_result, dry_run=dry_run)
    elif daily_digest_enabled():
        from notifications.digest_snapshot.digest_candidates import cycle_evidence
        evidence = {**dict(getattr(cycle_result, "allocation", {}) or {}), **execution}
        evidence.setdefault("execution_status", "completed" if broker_submission_done else "no_action")
        evidence["orders_pending_count"] = len(pending_orders)
        summary["notification_digest"] = cycle_evidence(evidence,
            tuple(getattr(cycle_result, "submitted_orders", ()) or ()), dry_run=dry_run)
    return summary


def _build_notification_delivery_log_for_report(
    *,
    platform: str,
    strategy_profile: str,
    run_id: str,
    dry_run: bool,
    orders_previewed_count: int,
    delivery_events: list[dict],
) -> dict:
    events = [dict(event) for event in delivery_events if dict(event).get("delivery_status") == "sent"]
    if not dry_run or orders_previewed_count <= 0 or not events:
        return {}
    return {
        "notification_schema_version": "hk_live_enablement_notification.v1",
        "notification_event_type": "hk_snapshot_live_enablement_dry_run",
        "notification_correlation_id": str(run_id or ""),
        "locales": ["en", "zh-Hans"],
        "profile": str(strategy_profile or ""),
        "platform": str(platform or ""),
        "validation_status": "passed",
        "orders_previewed": int(orders_previewed_count),
        "delivery_events": events,
        "notification_contains_profile": True,
        "notification_contains_platform": True,
        "notification_contains_validation_status": True,
        "notification_contains_order_preview_summary": True,
        "notification_redacts_sensitive_fields": True,
        "redaction_policy": "raw notification text is not persisted; only sha256 and length are recorded",
    }


def _build_notification_delivery_summary(delivery_events: list[dict]) -> dict:
    safe_fields = (
        "sink",
        "delivery_status",
        "transport_acknowledged",
        "error_type",
        "compact_text_sha256",
        "compact_text_length",
    )
    events = [
        {key: event[key] for key in safe_fields if key in event}
        for event in (dict(item) for item in delivery_events)
    ]
    if not events:
        return {}
    sent_count = sum(event.get("delivery_status") == "sent" for event in events)
    failed_count = sum(event.get("delivery_status") == "failed" for event in events)
    return {
        "attempted_count": len(events),
        "sent_count": sent_count,
        "failed_count": failed_count,
        "all_acknowledged": failed_count == 0 and sent_count == len(events),
        "delivery_events": events,
    }


signal_text = build_signal_text(t)
strategy_display_name = build_strategy_display_name(t)(
    STRATEGY_PROFILE,
    fallback_name=STRATEGY_DISPLAY_NAME,
    metadata=RUNTIME_SETTINGS.strategy_metadata,
    locale=NOTIFY_LANG,
)


def log_position_snapshot(message):
    print(f"[{ACCOUNT_REGION}] {message}", flush=True)


def log_runtime_warning(message):
    print(f"[{ACCOUNT_REGION}] [warning] {message}", flush=True)


BROKER_ADAPTERS = build_runtime_broker_adapters(
    strategy_symbols=tuple(MANAGED_SYMBOLS),
    account_hash=ACCOUNT_PREFIX or ACCOUNT_REGION or "longbridge",
    fetch_last_price_fn=fetch_last_price,
    fetch_strategy_account_state_fn=lambda quote_context, trade_context: fetch_strategy_account_state(
        quote_context,
        trade_context,
        list(MANAGED_SYMBOLS),
        cash_currency=TRADING_CURRENCY,
        position_log_fn=(
            log_position_snapshot
            if getattr(RUNTIME_SETTINGS, "debug_position_snapshot", False)
            else None
        ),
        warning_log_fn=log_runtime_warning,
    ),
    submit_order_fn=_profile_submit_order,
    symbol_suffix=SYMBOL_SUFFIX,
    currency=TRADING_CURRENCY,
    cash_only_execution=CASH_ONLY_EXECUTION,
)
STRATEGY_ADAPTERS = build_runtime_strategy_adapters(
    strategy_runtime=STRATEGY_RUNTIME,
    strategy_profile=STRATEGY_PROFILE,
    strategy_runtime_config=STRATEGY_RUNTIME_CONFIG,
    available_inputs=AVAILABLE_INPUTS,
    benchmark_symbol=BENCHMARK_SYMBOL,
    signal_text_fn=signal_text,
    translator=t,
    broker_adapters=BROKER_ADAPTERS,
    calculate_rotation_indicators_fn=calculate_rotation_indicators,
    build_strategy_evaluation_inputs_fn=build_strategy_evaluation_inputs,
    map_strategy_decision_to_plan_fn=map_strategy_decision_to_plan,
    execution_policy={
        "reserved_cash_floor_usd": getattr(RUNTIME_SETTINGS, "reserved_cash_floor_usd", 0.0),
        "reserved_cash_ratio": getattr(RUNTIME_SETTINGS, "reserved_cash_ratio", 0.0),
        "cash_only_execution": CASH_ONLY_EXECUTION,
    },
    execution_materials=getattr(STRATEGY_RUNTIME, "execution_materials", {}),
    build_strategy_plugin_report_payload_fn=build_strategy_plugin_report_payload,
    load_configured_strategy_plugin_signals_fn=load_configured_strategy_plugin_signals,
    parse_strategy_plugin_mounts_fn=parse_strategy_plugin_mounts,
)


def _safe_haven_cash_substitute_threshold_usd() -> float:
    return float(
        getattr(
            RUNTIME_SETTINGS,
            "safe_haven_cash_substitute_threshold_usd",
            DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD,
        )
    )


def _min_order_notional_usd() -> float:
    return float(getattr(RUNTIME_SETTINGS, "min_order_notional_usd", DEFAULT_MIN_ORDER_NOTIONAL_USD))


def build_composer(*, dry_run_only_override: bool | None = None):
    return build_runtime_composer(
        project_id=PROJECT_ID,
        secret_name=SECRET_NAME,
        token_refresh_threshold_days=TOKEN_REFRESH_THRESHOLD_DAYS,
        account_prefix=ACCOUNT_PREFIX,
        account_region=ACCOUNT_REGION,
        strategy_profile=STRATEGY_PROFILE,
        strategy_display_name=STRATEGY_DISPLAY_NAME,
        strategy_display_name_localized=strategy_display_name,
        strategy_domain=RUNTIME_SETTINGS.strategy_domain,
        notify_lang=NOTIFY_LANG,
        tg_token=TG_TOKEN,
        tg_chat_id=TG_CHAT_ID,
        notification_channel=_NOTIFICATION_CHANNEL,
        webhook_url=_NOTIFICATION_WEBHOOK_URL,
        managed_symbols=tuple(MANAGED_SYMBOLS),
        benchmark_symbol=BENCHMARK_SYMBOL,
        signal_effective_after_trading_days=SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS,
        separator=SEPARATOR,
        limit_sell_discount=LIMIT_SELL_DISCOUNT,
        limit_buy_premium=LIMIT_BUY_PREMIUM,
        limit_buy_premium_by_symbol=LIMIT_BUY_PREMIUM_BY_SYMBOL,
        order_poll_interval_sec=ORDER_POLL_INTERVAL_SEC,
        order_poll_max_attempts=ORDER_POLL_MAX_ATTEMPTS,
        safe_haven_cash_substitute_threshold_usd=_safe_haven_cash_substitute_threshold_usd(),
        min_order_notional_usd=_min_order_notional_usd(),
        market=MARKET,
        symbol_suffix=SYMBOL_SUFFIX,
        trading_currency=TRADING_CURRENCY,
        dry_run_only=RUNTIME_SETTINGS.dry_run_only,
        dry_run_only_override=dry_run_only_override,
        broker_adapters=BROKER_ADAPTERS,
        strategy_adapters=STRATEGY_ADAPTERS,
        estimate_max_purchase_quantity_fn=_profile_estimate_max_purchase_quantity,
        fetch_order_status_fn=fetch_order_status,
        fetch_token_from_secret_fn=fetch_token_from_secret,
        fetch_token_with_metadata_fn=fetch_account_snapshot_token_with_metadata,
        refresh_token_if_needed_fn=refresh_token_if_needed,
        build_contexts_fn=build_contexts,
        run_id_builder=build_run_id,
        event_logger=emit_runtime_log,
        report_builder=build_runtime_report_base,
        report_persister=persist_runtime_report,
        translator=t,
        runtime_target=RUNTIME_SETTINGS.runtime_target,
        env_reader=os.getenv,
        sleeper=time.sleep,
        printer=print,
        extra_reporting_fields={
            "market": MARKET,
            "market_calendar": MARKET_CALENDAR,
            "market_timezone": MARKET_TIMEZONE,
            "symbol_suffix": SYMBOL_SUFFIX,
            "trading_currency": TRADING_CURRENCY,
        },
    )


def build_strategy_plugin_alert_messages(signals):
    return STRATEGY_ADAPTERS.build_strategy_plugin_alert_messages(signals)


def _runtime_error_notification_targets() -> tuple[tuple[str, str], ...]:
    targets: list[tuple[str, str]] = []
    if TG_TOKEN and TG_CHAT_ID:
        targets.append((TG_TOKEN, TG_CHAT_ID))

    seen: set[tuple[str, str]] = set()
    unique_targets: list[tuple[str, str]] = []
    for target in targets:
        if target in seen:
            continue
        seen.add(target)
        unique_targets.append(target)
    return tuple(unique_targets)


def _runtime_error_notification_message(exc: Exception, *, route_label: str) -> str:
    title = "runtime_failure_title"
    return "\n".join((
        t(title),
        t("strategy_label", name=strategy_display_name or STRATEGY_PROFILE),
        t("runtime_failure_context", context=ACCOUNT_REGION),
        t("runtime_failure_result"),
        t("runtime_failure_action"),
    ))


def _send_runtime_error(exc: Exception, *, route_label: str) -> bool:
    targets = _runtime_error_notification_targets()
    if not targets:
        print(t("runtime_notification_missing_target"), flush=True)
        return False
    message = _runtime_error_notification_message(exc, route_label=route_label)
    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)
        except Exception:
            print(
                t("runtime_notification_delivery_failed"),
                flush=True,
            )
            outcomes.append(False)
    return bool(outcomes) and all(outcomes)


def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
    return unified_notifications.dispatch_immediate(
        lambda: _send_runtime_error(exc, route_label=route_label), kind="real_error",
        incident=f"runtime_route:{route_label}:{_runtime_failure_reason(exc)}",
        occurred_at=datetime.now(timezone.utc), run_id=build_run_id(),
    ) if unified_notifications.enabled() else _send_runtime_error(exc, route_label=route_label)


def _handle_route_runtime_error(exc: Exception, *, route_label: str):
    print(t("runtime_failure_log"), flush=True)
    _notify_runtime_error(exc, route_label=route_label)
    return "Error", 500


def _route_with_runtime_error_fallback(handler, *, success_body: str, route_label: str, **kwargs):
    try:
        if unified_notifications.enabled():
            with unified_notifications.bind_route(route_label):
                result = handler(**kwargs)
        else:
            result = handler(**kwargs)
    except Exception as exc:
        return _handle_route_runtime_error(exc, route_label=route_label)
    if result is False:
        return "Error", 500
    if isinstance(result, tuple):
        return result
    return success_body, 200


def build_strategy_plugin_alert_state_settings():
    return StrategyPluginAlertStateSettings.from_env(
        project_id=PROJECT_ID,
    )


def build_strategy_plugin_alert_context_label() -> str:
    return build_alert_context_label(
        platform_id="longbridge",
        strategy_profile=STRATEGY_PROFILE,
        account_scope=ACCOUNT_REGION,
        service_name=SECRET_NAME,
        runtime_target=RUNTIME_SETTINGS.runtime_target,
    )


def publish_strategy_plugin_alerts(signals, *, report=None):
    from notifications.digest_snapshot.digest_routing import immediate_plugin_signals
    result = dispatch_strategy_plugin_alerts(
        immediate_plugin_signals(signals),
        notification_settings=RUNTIME_SETTINGS,
        translator=t,
        strategy_label=STRATEGY_PROFILE,
        context_label=build_strategy_plugin_alert_context_label(),
        state_settings=build_strategy_plugin_alert_state_settings(),
        log_message=print,
    )
    if report is not None:
        result.attach_to_report(report)
    return result


def run_strategy(*, force_run: bool = False, validation_only: bool = False, validation_label: str = "backfill"):
    if _v7_application_candidate_bound() and not validation_only:
        print(f"[{datetime.now()}] V7 paper candidate is paused; skip all execution paths.", flush=True)
        return True
    if not validation_only and not force_run and not getattr(RUNTIME_SETTINGS, "runtime_target_enabled", True):
        print(f"[{datetime.now()}] Runtime target disabled; skip strategy execution.", flush=True)
        return True
    composer = build_composer(dry_run_only_override=True if validation_only else None)
    reporting_adapters = composer.build_reporting_adapters()
    log_context, report = reporting_adapters.start_run()
    notification_delivery_events: list[dict] = []
    try:
        notification_adapters = composer.build_notification_adapters(
            delivery_events=notification_delivery_events,
        )
    except TypeError as exc:
        if "delivery_events" not in str(exc):
            raise
        notification_adapters = composer.build_notification_adapters()
    strategy_plugin_signals, strategy_plugin_error = composer.load_strategy_plugin_signals(
        getattr(RUNTIME_SETTINGS, "strategy_plugin_mounts_json", None)
    )
    composer.attach_strategy_plugin_report(
        report,
        signals=strategy_plugin_signals,
        error=strategy_plugin_error,
    )
    failure_phase = "market_hours"
    try:
        reporting_adapters.log_event(
            log_context,
            "strategy_cycle_started",
            message="Starting strategy execution",
        )
        print(composer.with_prefix(f"[{datetime.now()}] Starting strategy..."), flush=True)

        market_open = is_market_open_now(
            calendar_name=MARKET_CALENDAR,
            timezone_name=MARKET_TIMEZONE,
        )
        if isinstance(market_open, tuple):
            market_open, error = market_open
            if error is not None:
                reporting_adapters.log_event(
                    log_context,
                    "market_hours_check_failed",
                    message="Market hours check failed",
                    severity="WARNING",
                    error_message=str(error),
                    market=MARKET,
                    market_calendar=MARKET_CALENDAR,
                    market_timezone=MARKET_TIMEZONE,
                )
                print(composer.with_prefix(f"Market hours check failed: {error}"), flush=True)
        if not market_open and not force_run:
            reporting_adapters.log_event(
                log_context,
                "outside_market_hours",
                message="Outside market hours; skip execution",
                market=MARKET,
                market_calendar=MARKET_CALENDAR,
                market_timezone=MARKET_TIMEZONE,
            )
            finalize_runtime_report(
                report,
                status="skipped",
                diagnostics={
                    "skip_reason": "market_closed",
                },
            )
            print(composer.with_prefix("Outside market hours; skip."), flush=True)
            return True
        if force_run and not market_open:
            reporting_adapters.log_event(
                log_context,
                "market_hours_bypassed",
                message=f"Market hours bypassed for {validation_label} execution",
                market=MARKET,
                market_calendar=MARKET_CALENDAR,
                market_timezone=MARKET_TIMEZONE,
            )
            print(
                composer.with_prefix(
                    f"Market hours bypassed for {validation_label} verification; validation only, no orders will be submitted."
                ),
                flush=True,
            )
        unsupported_reason = dca_execution_unsupported_reason(STRATEGY_PROFILE)
        if unsupported_reason is not None:
            reporting_adapters.log_event(
                log_context,
                "strategy_execution_unsupported",
                message="Strategy requires fractional-share execution; skip",
                skip_reason=unsupported_reason,
                strategy_profile=STRATEGY_PROFILE,
                market=MARKET,
                market_calendar=MARKET_CALENDAR,
                market_timezone=MARKET_TIMEZONE,
            )
            finalize_runtime_report(
                report,
                status="skipped",
                diagnostics={"skip_reason": unsupported_reason},
            )
            print(
                composer.with_prefix(
                    f"Strategy {STRATEGY_PROFILE} requires fractional-share execution; skip."
                ),
                flush=True,
            )
            return True
        if not validation_only:
            publish_strategy_plugin_alerts(strategy_plugin_signals, report=report)
        failure_phase = "runtime_configuration"
        try:
            rebalance_runtime = composer.build_rebalance_runtime(
                silent_cycle_notifications=validation_only,
                notification_delivery_events=notification_delivery_events,
            )
        except TypeError as exc:
            if "notification_delivery_events" not in str(exc):
                raise
            rebalance_runtime = composer.build_rebalance_runtime(
                silent_cycle_notifications=validation_only,
            )
        rebalance_config = composer.build_rebalance_config(
            strategy_plugin_signals=strategy_plugin_signals,
            strategy_plugin_error=strategy_plugin_error,
            notification_title_key=(
                "dry_run_title"
                if validation_only and validation_label == "dry_run"
                else ""
            ),
            cash_only_execution=CASH_ONLY_EXECUTION,
            live_execution_session_authorized=bool(market_open and not validation_only),
        )
        failure_phase = "strategy_cycle"
        if unified_notifications.enabled():
            with unified_notifications.bind_report(report):
                cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config)
        else:
            cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config)
        failure_phase = "reporting"
        signal_snapshot = {}
        if cycle_result is not None:
            execution = dict(getattr(cycle_result, "execution", {}) or {})
            signal_snapshot = dict(execution.get("signal_snapshot") or {})
        execution_summary = _summarize_cycle_result_for_report(
            cycle_result,
            dry_run=bool(report.get("dry_run")),
        )
        if unified_notifications.enabled():
            try:
                unified_notifications.attach_cycle_event(execution_summary, cycle_result, report=report, config=rebalance_config)
            except Exception:
                print("notification_event_attachment_unconfirmed", flush=True)
                unified_notifications.record_failure("notification_event_attachment_unconfirmed", report=report)
        notification_delivery_log = _build_notification_delivery_log_for_report(
            platform="longbridge",
            strategy_profile=STRATEGY_PROFILE,
            run_id=str(report.get("run_id") or ""),
            dry_run=bool(report.get("dry_run")),
            orders_previewed_count=int(execution_summary.get("orders_previewed_count") or 0),
            delivery_events=notification_delivery_events,
        )
        if notification_delivery_log:
            execution_summary["notification_delivery_log"] = notification_delivery_log
        if unified_notifications.enabled():
            notification_delivery_events.extend(
                {"sink": "notification_store", "delivery_status": "failed", "transport_acknowledged": False,
                 "error_type": code} for code in report.get("notification_unified_errors", ())
            )
        notification_delivery_summary = _build_notification_delivery_summary(
            notification_delivery_events
        )
        if notification_delivery_summary:
            execution_summary["notification_delivery_summary"] = notification_delivery_summary
        if signal_snapshot:
            reporting_adapters.log_event(
                log_context,
                "strategy_signal_snapshot",
                message="Strategy signal snapshot",
                **signal_snapshot,
            )
        try:
            attach_cycle_execution_receipt(report, cycle_result)
        except ValueError:
            # Test/legacy runtime reports without a release attestation remain
            # evidence-missing; receipt metadata must not change a completed
            # strategy cycle into a failure.
            pass
        finalize_runtime_report(
            report,
            status="ok",
            summary=execution_summary,
            diagnostics={"signal_snapshot": signal_snapshot} if signal_snapshot else None,
        )
        reporting_adapters.log_event(
            log_context,
            "strategy_cycle_completed",
            message="Strategy execution completed",
        )
        return True

    except Exception as exc:
        failure_reason = _runtime_failure_reason(exc)
        append_runtime_report_error(
            report,
            stage="strategy_cycle",
            message="strategy_cycle_failed",
            error_type=type(exc).__name__,
        )
        reporting_adapters.log_event(
            log_context,
            "strategy_cycle_failed",
            message="Strategy execution failed",
            severity="ERROR",
            error_type=type(exc).__name__,
            error_message="strategy_cycle_failed",
            failure_phase=failure_phase,
            failure_reason=failure_reason,
        )
        if not (validation_only and _v7_application_candidate_bound()):
            try:
                def publish_failure():
                    return notification_adapters.publish_cycle_notification(
                        detailed_text=_compact_error_notification(exc, phase=failure_phase),
                        compact_text=_compact_error_notification(exc, phase=failure_phase),
                    )
                unified_notifications.dispatch_immediate(publish_failure, kind="real_error",
                    incident=f"runtime_failure:{failure_phase}:{failure_reason}", report=report, actually_ran=True)
            except Exception as notification_exc:
                notification_delivery_events.append(
                    {
                        "sink": "telegram",
                        "delivery_status": "failed",
                        "transport_acknowledged": False,
                        "error_type": type(notification_exc).__name__,
                    }
                )
                reporting_adapters.log_event(
                    log_context,
                    "strategy_error_notification_failed",
                    message="Strategy error notification failed",
                    severity="ERROR",
                    error_type=type(notification_exc).__name__,
                )
        error_summary = {}
        notification_delivery_summary = _build_notification_delivery_summary(
            notification_delivery_events
        )
        if notification_delivery_summary:
            error_summary["notification_delivery_summary"] = (
                notification_delivery_summary
            )
        finalize_runtime_report(
            report,
            status="error",
            summary=error_summary or None,
            diagnostics={"failure_phase": failure_phase, "failure_reason": failure_reason},
        )
        return False
    finally:
        if "execution_receipt" not in report:
            try:
                attach_terminal_fallback_execution_receipt(report)
            except ValueError:
                # Legacy targets remain evidence-missing; receipt metadata
                # cannot affect strategy execution or error handling.
                pass
        if not (validation_only and _v7_application_candidate_bound()):
            try:
                report_path = reporting_adapters.persist_execution_report(report)
                print(f"execution_report {report_path}", flush=True)
                if unified_notifications.enabled():
                    unified_notifications.recover_runtime(report=report)
            except Exception as persist_exc:
                print(f"failed to persist execution report: {type(persist_exc).__name__}", flush=True)


def _bounded_probe_error_code(exc: Exception) -> int | None:
    try:
        code = getattr(exc, "code", None)
    except Exception:
        return None
    if type(code) is not int or not -(2**31) <= code <= 2**31 - 1:
        return None
    return code


def run_probe(*, response_body: str = "Probe OK"):
    if _v7_application_candidate_bound():
        return response_body, 200
    history_scope = str(ACCOUNT_REGION or "").strip()
    record_account_history = (
        history_scope in {"HK", "SG"}
        and str(os.getenv("ACCOUNT_HISTORY_RECORDING_ENABLED") or "").strip() == "true"
    )
    history_capture = begin_natural_cycle_history() if record_account_history else None
    composer = None
    reporting_adapters = None
    log_context = None
    report = None
    probe_step = "settings"
    try:
        composer = build_composer(dry_run_only_override=True)
        reporting_adapters = composer.build_reporting_adapters()
        log_context, report = reporting_adapters.start_run()
        reporting_adapters.log_event(
            log_context,
            "health_probe_received",
            message="Received health probe request",
            execution_window="probe",
        )
        probe_step = "context"
        runtime = composer.build_rebalance_runtime(silent_cycle_notifications=True)
        if record_account_history:
            probe_step = "context"
            quote_context, trade_context, source_binding = composer.build_account_snapshot_broker_contexts()
            probe_step = "balance"
            read_cycle_account_balance(lambda: trade_context.account_balance(currency="USD"))
            observation = cycle_history_observation()
            if observation is None:
                raise RuntimeError("account history observation incomplete")
            probe_step = "archive"
            result = record_projected_daily_account(
                os.environ,
                account_scope=history_scope,
                source_binding=source_binding,
                balances=observation["projection"]["broker_reported_balances"],
                cash=observation["projection"]["cash"],
                started=observation["started"],
                finished=observation["finished"],
                open_store=_open_account_history_store,
                financing=observation["projection"].get("financing"),
            )
            print(f"account_history status={result.status} category={result.category or '-'}", flush=True)
            if result.status != "recorded":
                raise RuntimeError("account history write was not confirmed")
            probe_step = "indicators"
            _indicators = composer.strategy_adapters.calculate_strategy_indicators(quote_context)
        else:
            probe_step = "bootstrap"
            quote_context, trade_context, _indicators = runtime.bootstrap()
        probe_step = "portfolio_snapshot"
        snapshot = runtime.portfolio_port_factory(
            quote_context,
            trade_context,
        ).get_portfolio_snapshot()
        positions = tuple(getattr(snapshot, "positions", ()) or ())
        buying_power = float(getattr(snapshot, "buying_power", 0.0) or 0.0)
        total_equity = float(getattr(snapshot, "total_equity", 0.0) or 0.0)
        finalize_runtime_report(
            report,
            status="ok",
            summary={
                "buying_power": buying_power,
                "total_equity": total_equity,
                "positions_count": len(positions),
            },
        )
        reporting_adapters.log_event(
            log_context,
            "health_probe_completed",
            message="Health probe completed",
            execution_window="probe",
            buying_power=buying_power,
            total_equity=total_equity,
            positions_count=len(positions),
        )
        return response_body, 200
    except Exception as exc:
        error_code = _bounded_probe_error_code(exc)
        error_diagnostics = {
            "probe_step": probe_step,
            "error_code_status": "known" if error_code is not None else "unknown",
        }
        if error_code is not None:
            error_diagnostics["error_code"] = error_code
        if report is not None:
            append_runtime_report_error(
                report,
                stage="health_probe",
                message="health_probe_failed",
                error_type=type(exc).__name__,
                **error_diagnostics,
            )
            finalize_runtime_report(report, status="error")
        if reporting_adapters is not None and log_context is not None:
            reporting_adapters.log_event(
                log_context,
                "health_probe_failed",
                message="Health probe failed",
                severity="ERROR",
                execution_window="probe",
                error_type=type(exc).__name__,
                error_message="health_probe_failed",
                **error_diagnostics,
            )
        err = _compact_error_notification(
            exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"), phase="health_probe",
        )
        if composer is not None:
            def publish_probe_failure():
                return composer.build_notification_adapters().publish_cycle_notification(
                    detailed_text=err,
                    compact_text=_compact_error_notification(
                        exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"), phase="health_probe",
                    ),
                )
            unified_notifications.dispatch_immediate(publish_probe_failure, kind="real_error",
                incident="runtime_failure:health_probe", report=report)
        else:
            print(err, flush=True)
        return "Error", 500
    finally:
        if history_capture is not None:
            end_natural_cycle_history(history_capture)
        try:
            if reporting_adapters is not None and report is not None:
                report_path = reporting_adapters.persist_execution_report(report)
                print(f"execution_report {report_path}", flush=True)
                if unified_notifications.enabled():
                    unified_notifications.recover_runtime(report=report, probe=True)
        except Exception as persist_exc:
            print(f"failed to persist execution report: {type(persist_exc).__name__}", flush=True)


def _paper_command_consumer_session_date() -> str:
    """Return the exchange-local session date used to claim due commands."""
    try:
        import pytz

        return datetime.now(pytz.timezone(MARKET_TIMEZONE)).date().isoformat()
    except Exception:
        return datetime.now().date().isoformat()


def _paper_command_consumer_runtime_is_isolated() -> bool:
    runtime_target = getattr(RUNTIME_SETTINGS, "runtime_target", None)
    return bool(
        getattr(RUNTIME_SETTINGS, "dry_run_only", False)
        and runtime_target is not None
        and str(getattr(runtime_target, "execution_mode", "") or "").lower() == "paper"
    )


def run_paper_execution_command_consumer() -> bool:
    """Run an explicitly invoked, read-only paper command consumer.

    This handler never constructs an execution port or calls the LongBridge
    order API.  It is deliberately separate from ``/run`` so a normal cycle
    cannot start consuming delayed commands by accident.
    """
    if _v7_application_candidate_bound():
        return True
    if not _paper_command_consumer_runtime_is_isolated():
        raise RuntimeError(
            "paper command consumer requires RUNTIME_TARGET_JSON.execution_mode=paper "
            "and LONGBRIDGE_DRY_RUN_ONLY=true"
        )
    if not resolve_paper_execution_command_consumer_enabled(
        env_reader=os.getenv,
        dry_run_only=bool(RUNTIME_SETTINGS.dry_run_only),
    ):
        raise RuntimeError("paper command consumer is not enabled")

    composer = build_composer()
    config = composer.build_rebalance_config()
    reporting_adapters = composer.build_reporting_adapters()
    log_context, report = reporting_adapters.start_run()
    try:
        reporting_adapters.log_event(
            log_context,
            "paper_execution_command_consumer_started",
            message="Starting read-only paper execution command consumer",
        )
        quote_context, trade_context = composer.build_read_only_broker_contexts()
        portfolio = composer.broker_adapters.build_portfolio_port(
            quote_context,
            trade_context,
        ).get_portfolio_snapshot()
        result = consume_due_paper_execution_commands(
            store=config.execution_command_store,
            as_of_session=_paper_command_consumer_session_date(),
            claimant=str(os.getenv("K_SERVICE") or "longbridge-paper-command-consumer"),
            portfolio=portfolio,
            market_data_port=composer.broker_adapters.build_market_data_port(quote_context),
            runtime_release_receipt=config.runtime_release_receipt,
            expected_strategy_release=config.expected_strategy_release,
            expected_command_binding={
                "platform": "longbridge",
                "account_scope": str(config.execution_state_account_scope or "unknown"),
                "strategy_profile": str(config.strategy_profile or "unknown"),
            },
        )
        report_status = "ok" if result.get("status") == "ok" else "skipped"
        finalize_runtime_report(
            report,
            status=report_status,
            summary={"paper_execution_command_consumer": result},
        )
        reporting_adapters.log_event(
            log_context,
            "paper_execution_command_consumer_completed",
            message="Paper execution command consumer completed",
            result_status=result.get("status"),
            commands_count=len(tuple(result.get("commands") or ())),
        )
        return True
    except Exception as exc:
        append_runtime_report_error(
            report,
            stage="paper_execution_command_consumer",
            message=str(exc),
            error_type=type(exc).__name__,
        )
        finalize_runtime_report(report, status="error")
        reporting_adapters.log_event(
            log_context,
            "paper_execution_command_consumer_failed",
            message="Paper execution command consumer failed",
            severity="ERROR",
            error_type=type(exc).__name__,
            error_message=str(exc),
        )
        return False
    finally:
        try:
            report_path = reporting_adapters.persist_execution_report(report)
            print(f"execution_report {report_path}", flush=True)
        except Exception as persist_exc:
            print(f"failed to persist execution report: {persist_exc}", flush=True)


def run_broker_reconciliation():
    """Run the explicitly enabled, order-port-free reconciliation boundary."""

    return run_read_only_broker_reconciliation(
        enabled=reconciliation_enabled(os.getenv),
        account_scope=getattr(RUNTIME_SETTINGS, "account_region", None),
        runtime_target=getattr(RUNTIME_SETTINGS, "runtime_target", None),
        strategy_profile=STRATEGY_PROFILE,
        project_id=PROJECT_ID,
        build_read_only_contexts=lambda: build_composer().build_read_only_broker_contexts(),
        collect_evidence=READ_ONLY_BROKER_RECONCILIATION_COLLECTOR,
    )


def fetch_account_snapshot_token_with_metadata(project_id, secret_name):
    """Read one configured secret-store response. A metadata failure is not reread."""
    from types import SimpleNamespace

    from quant_platform_kit.cloud import get_secret_store

    store = get_secret_store()
    read_metadata = getattr(store, "get_secret_with_metadata", None)
    if callable(read_metadata):
        return read_metadata(secret_name, project_id=project_id)
    return SimpleNamespace(
        value=store.get_secret(secret_name, project_id=project_id),
        version_name=None,
    )


def _open_account_history_store(project_id):
    from quant_platform_kit.cloud import get_object_store

    return get_object_store(project_id=project_id)


def run_account_snapshot():
    """Read bounded account facts through the order-port-free broker contexts."""

    return run_read_only_account_snapshot(
        enabled=account_snapshot_enabled(os.getenv),
        account_scope=getattr(RUNTIME_SETTINGS, "account_region", None),
        build_read_only_contexts=lambda: build_composer().build_account_snapshot_broker_contexts(),
        collect_evidence=READ_ONLY_BROKER_RECONCILIATION_COLLECTOR,
    )


@app.route("/run", methods=["POST"])
def handle_trigger():
    """Entrypoint for Cloud Run / scheduler: run strategy and return 200."""
    if request_method() != "POST":
        return "Method Not Allowed", 405
    return _route_with_runtime_error_fallback(
        run_strategy,
        success_body="OK",
        route_label="POST /run",
    )


@app.route("/backfill", methods=["POST", "GET"])
def handle_backfill():
    """Manual backfill entrypoint for verification-only execution."""
    return _route_with_runtime_error_fallback(
        run_strategy,
        force_run=True,
        validation_only=True,
        success_body="OK",
        route_label="POST /backfill",
    )


@app.route("/dry-run", methods=["POST", "GET"])
def handle_dry_run():
    """Strategy dry-run entrypoint."""
    return _route_with_runtime_error_fallback(
        run_strategy,
        force_run=True,
        validation_only=True,
        validation_label="dry_run",
        success_body="Dry Run OK",
        route_label="POST /dry-run",
    )


@app.route("/probe", methods=["POST"])
def handle_probe():
    """Post-open broker/account health probe; notify only on failure."""
    if request_method() != "POST":
        return "Method Not Allowed", 405
    return _route_with_runtime_error_fallback(
        run_probe,
        success_body="Probe OK",
        route_label="POST /probe",
    )


@app.route("/paper-command-consumer", methods=["POST"])
def handle_paper_execution_command_consumer():
    """Explicit paper-only command verification endpoint; never scheduler-routed."""
    return _route_with_runtime_error_fallback(
        run_paper_execution_command_consumer,
        success_body="Paper command consumer OK",
        route_label="POST /paper-command-consumer",
    )


@app.route("/reconcile", methods=["POST"])
def handle_broker_reconciliation():
    """Return only redacted QPK E3 evidence; never invoke normal execution."""

    if request_method() != "POST":
        return "Method Not Allowed", 405
    payload, status = run_broker_reconciliation()
    return json.dumps(payload, ensure_ascii=True), status, {"Content-Type": "application/json"}


@app.route("/account-snapshot", methods=["GET"])
def handle_account_snapshot():
    """Return a non-cacheable, read-only and authority-free account snapshot."""

    if request_method() != "GET":
        return "Method Not Allowed", 405
    payload, status = run_account_snapshot()
    return json.dumps(payload, ensure_ascii=True), status, {
        "Content-Type": "application/json",
        "Cache-Control": "no-store",
    }


@app.route("/v7-paper-application-loaded", methods=["GET"])
def handle_v7_paper_application_loaded():
    """Read back the disabled V7 process identity without broker access."""

    if request_method() != "GET":
        return "Method Not Allowed", 405
    from application.v7_paper_application import (
        V7PaperApplicationError,
        build_v7_loaded_version_readback,
        load_v7_paper_application_binding,
    )
    try:
        binding = load_v7_paper_application_binding(os.environ)
        if binding is None:
            return json.dumps(
                {"status": "unavailable", "reason": "application_binding_missing"},
                ensure_ascii=True,
            ), 409, {"Content-Type": "application/json", "Cache-Control": "no-store"}
        payload = build_v7_loaded_version_readback(
            binding,
            runtime_settings=RUNTIME_SETTINGS,
            strategy_runtime=STRATEGY_RUNTIME,
            runtime_env=os.environ,
        )
    except V7PaperApplicationError:
        return json.dumps(
            {"status": "uncertain", "reason": "loaded_version_unavailable"},
            ensure_ascii=True,
        ), 409, {"Content-Type": "application/json", "Cache-Control": "no-store"}
    return json.dumps(
        {"status": "applied_paused", "readback": payload},
        ensure_ascii=True,
    ), 200, {"Content-Type": "application/json", "Cache-Control": "no-store"}


def _v7_application_candidate_bound() -> bool:
    """Keep the exact V7 binding paused even when a route bypasses the flag."""

    if STRATEGY_PROFILE != "soxl_soxx_core_only_p2_v7_longterm_compounding_cash_reserve":
        return False
    from application.v7_paper_application import load_v7_paper_application_binding

    return load_v7_paper_application_binding(os.environ) is not None


@app.route("/monitor-dispatch", methods=["POST", "GET"])
def handle_monitor_dispatch():
    if request_method() == "GET":
        return "Monitor Dispatch OK - use POST to dispatch due monitor checks", 200
    try:
        result = dispatch_due_monitors(_load_monitor_dispatch_targets())
    except Exception as exc:
        return _handle_route_runtime_error(exc, route_label="POST /monitor-dispatch")
    return json.dumps(result, ensure_ascii=False), 200, {"Content-Type": "application/json"}


def request_method() -> str:
    try:
        return request.method
    except Exception:
        return "GET"


@app.route("/healthz", methods=["GET"])
@app.route("/health", methods=["GET"])
def health():
    critical_errors: list[str] = []
    for module_path in (
        "application.runtime_composer",
        "application.runtime_reporting_adapters",
        "application.runtime_strategy_adapters",
        "runtime_config_support",
        "runtime_execution_policy",
    ):
        try:
            importlib.import_module(module_path)
        except Exception as exc:
            critical_errors.append(f"{module_path}: {type(exc).__name__}: {exc}")
    if critical_errors:
        return json.dumps({"status": "unhealthy", "errors": critical_errors}, ensure_ascii=False), 500
    return "OK", 200


if __name__ == "__main__":
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))