import argparse
import fcntl
import hashlib
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import contextmanager
from copy import deepcopy
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional

# Support both `python src/pipeline.py` and `from src.pipeline import ...`
# when caller cwd is `bridge/` and top-level bare imports are used below.
_SRC_DIR = Path(__file__).resolve().parent
if str(_SRC_DIR) not in sys.path:
    sys.path.insert(0, str(_SRC_DIR))

import config_manager
from executor import run_engine
from execution_state import (
    apply_execution_report,
    create_or_get_signal_state,
    get_signal_state,
    is_final_state,
    transition_state,
    transition_states,
)
from incident_engine import evaluate_incidents, resolve_auto_actions
from live_volume import resolve_live_open_guard, resolve_live_volume_plan
try:
    from src.ledger import append_ledger_event
except ModuleNotFoundError:
    # Support direct script execution: `python3 src/pipeline.py`
    from ledger import append_ledger_event
from mt5_adapter import load_mt5_snapshot, to_engine_request
from mt5_execution import dispatch_execution
from ea_bridge_live import EABridgeLiveSource
from mt5_live import MT5LiveSource
from agent_decision_contract import load_agent_decision_contract
from runtime_state.decision_snapshot import (
    build_macro_direction_snapshot,
    resolve_allowed_side,
    resolve_macro_authority,
)
from strategy_intelligence import get_intelligence_hub
from tenant_manager import load_runtime_state, tenant_paths
from rollout_manager import check_rollout_safety, evaluate_promotion_gates
from metrics_agg import aggregate_exec_stats
try:
    from src.position_lifecycle import (
        normalize_position_attribution,
        sync_close_fill,
        sync_close_request,
        sync_open_fill,
        sync_open_request,
    )
except ModuleNotFoundError:
    from position_lifecycle import (
        normalize_position_attribution,
        sync_close_fill,
        sync_close_request,
        sync_open_fill,
        sync_open_request,
    )

ACK_RETRY_COOLDOWN_SECONDS = 20.0
TIMEOUT_RETRY_COOLDOWN_SECONDS = 30.0
PENDING_ENTRY_TTL_SECONDS = {
    "OPEN_LIMIT": 30 * 60,
    "ARM_WAIT_CONFIRMATION": 20 * 60,
}
MAX_TIMEOUT_RETRIES = 3
LIVE_SOURCE_RETRY_ERRORS = {
    "LIVE_SOURCE_SNAPSHOT_UNAVAILABLE",
    "LIVE_SOURCE_NOT_CONNECTED",
}
_RUNTIME_BLACKBOARD = None
_MAILBOX_LAST_EMIT: Dict[str, float] = {}
_PIPELINE_LOCK_HANDLE = None
APP_DIR_ENV = "LOBSTERQUANT_APP_DIR"
MACRO_FRESH_TTL_SECONDS = 20.0 * 60.0
MACRO_EXPIRE_TTL_SECONDS = 45.0 * 60.0
TRENDING_REGIMES = {
    "TREND",
    "TRENDING",
    "BREAKOUT",
    "EXTREME_TREND",
    "FLAG_CONSOLIDATION",
    "SHOCK_REPRICING",
    "NORMAL_TREND",
}
TRENDING_SUBTYPES = {
    "FLAG_COMPRESSION",
    "REPRICING_IMPULSE",
    "CONFIRM_BREAK",
    "TREND_FIRST_PULLBACK",
    "TREND_CHANNEL",
}
RANGING_REGIMES = {"RANGE", "RANGING", "WIDE_RANGE"}
RANGING_SUBTYPES = {
    "WIDE_BOX",
    "LIQUIDITY_SWEEP",
    "SPRING",
    "UPTHRUST",
    "REVERSAL_SETUP",
    "NARROW_BOX",
    "DRIFTING_CHANNEL",
}
STRUCTURAL_HARD_BLOCK_ADVICES = {"LIMIT_ONLY", "WAIT_CONFIRMATION", "BLOCK_COUNTER_TREND"}


@dataclass
class PipelineRunContext:
    project_root: Path
    code_root: Optional[Path]
    tenant_id: str
    source: str
    runtime_intent: str
    output_path: Path
    execution_mode: str
    symbol: str
    execution_report_path: Optional[Path]
    enable_incident_autofix: bool
    runtime_state: Dict[str, Any] = field(default_factory=dict)
    promotion_info: Dict[str, Any] = field(default_factory=dict)
    mt5_snapshot: Dict[str, Any] = field(default_factory=dict)
    engine_request: Dict[str, Any] = field(default_factory=dict)
    scheduler_snapshot: Dict[str, Any] = field(default_factory=dict)
    agent_context: Dict[str, Any] = field(default_factory=dict)
    strategy_signals: list[Dict[str, Any]] = field(default_factory=list)
    aggregated_intent: Dict[str, Any] = field(default_factory=dict)
    signal: Dict[str, Any] = field(default_factory=dict)
    signal_id: str = ""
    signal_attribution: Dict[str, Any] = field(default_factory=dict)
    state_path: Optional[Path] = None


@dataclass
class PipelineOutcome:
    execution_report: Dict[str, Any] = field(default_factory=dict)
    incidents: list[Dict[str, Any]] = field(default_factory=list)
    auto_actions: list[Dict[str, Any]] = field(default_factory=list)
    ledger_events: list[str] = field(default_factory=list)
    extra_fields: Dict[str, Any] = field(default_factory=dict)
    include_runtime: bool = True
    append_runtime_log: bool = False
    sync_mailbox: bool = False


@dataclass
class StrategyOpportunity:
    strategy_id: str = ""
    instance_id: str = ""
    directional_view: str = "NONE"
    setup_quality: str = "INVALID"
    entry_style_preference: str = "MARKET"
    long_opportunity_score: float = 0.0
    short_opportunity_score: float = 0.0
    urgency: str = "LOW"
    confidence: float = 0.0
    actionable: bool = False
    source_action: str = "HOLD"
    source_side: str = "NONE"
    notes: list[str] = field(default_factory=list)


@dataclass
class ExecutionConstraint:
    advice: str = "ALLOW_FOLLOW"
    allowed_side: str = "BOTH"
    allowed_entry_styles: list[str] = field(default_factory=lambda: ["MARKET", "LIMIT"])
    confirmation_required: bool = False
    probe_size_scale: float = 0.5
    structural_side: str = "NONE"
    macro_direction: str = "NEUTRAL"
    block_new_open: bool = False
    reason: str = ""


@dataclass
class MaterializedOpportunity:
    materialized_action: str = "NO_SETUP"
    final_action: str = "HOLD"
    final_side: str = "NONE"
    entry_style: str = "MARKET"
    size_scale: float = 1.0
    eligible_for_aggregation: bool = False
    reason: str = ""


def _to_bool(value: str) -> bool:
    return str(value).lower() in {"1", "true", "yes", "on"}


def _read_json_dict(path: Path) -> Dict[str, Any]:
    if not path.exists():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        return payload if isinstance(payload, dict) else {}
    except Exception:
        return {}


def _runtime_state_path(project_root: Path, tenant_id: str) -> Path:
    return tenant_paths(project_root, tenant_id).get("runtime", project_root / "ops/runtime_state.json")


def _save_runtime_state(project_root: Path, tenant_id: str, state: Dict[str, Any]) -> None:
    _write_json(_runtime_state_path(project_root, tenant_id), state)


def _dedupe_string_list(values: Any) -> list[str]:
    if not isinstance(values, list):
        return []
    items: list[str] = []
    for value in values:
        text = str(value or "").strip()
        if text and text not in items:
            items.append(text)
    return items


def _incident_controls(runtime_state: Dict[str, Any]) -> Dict[str, Any]:
    controls = runtime_state.get("incident_controls", {}) if isinstance(runtime_state.get("incident_controls", {}), dict) else {}
    return dict(controls)


def _parse_auto_action_step(step: str) -> tuple[str, str]:
    raw = str(step or "").strip()
    if not raw:
        return "", ""
    action, sep, value = raw.partition(":")
    if not sep:
        return action.strip().upper(), ""
    return action.strip().upper(), value.strip()


def _apply_auto_action_step(controls: Dict[str, Any], step: str, *, executed_at: str) -> tuple[bool, Optional[str]]:
    action, raw_value = _parse_auto_action_step(step)
    if not action:
        return False, None
    if action == "DISABLE_NEW_OPEN":
        controls["disable_new_open"] = _to_bool(raw_value or "true")
        controls["disable_new_open_reason"] = f"auto_action:{step}"
        controls["disable_new_open_updated_at"] = executed_at
        return True, None
    if action == "SET_RUNTIME_MODE" and raw_value:
        controls["runtime_mode"] = raw_value
        controls["runtime_mode_updated_at"] = executed_at
        return True, None
    if action == "PAGE_ONCALL":
        return True, "PAGE_ONCALL requested by incident auto-action" if _to_bool(raw_value or "true") else None
    if action == "EXECUTOR_RECONCILE":
        controls["executor_reconcile_requested"] = _to_bool(raw_value or "true")
        controls["executor_reconcile_updated_at"] = executed_at
        return True, None
    if action == "RETRY_WITH_BACKOFF":
        controls["retry_with_backoff_requested"] = _to_bool(raw_value or "true")
        controls["retry_with_backoff_updated_at"] = executed_at
        return True, None
    if action == "RAISE_SPREAD_FILTER":
        try:
            controls["spread_filter_multiplier"] = float(raw_value)
            controls["spread_filter_updated_at"] = executed_at
            return True, None
        except Exception:
            return False, None
    if action == "REDUCE_MAX_RISK_PER_TRADE":
        try:
            controls["max_risk_per_trade_scale"] = float(raw_value)
            controls["max_risk_per_trade_updated_at"] = executed_at
            return True, None
        except Exception:
            return False, None
    return False, None


def _consume_pending_auto_actions(project_root: Path, tenant_id: str) -> tuple[Dict[str, Any], list[str], list[str], list[str]]:
    runtime_state = load_runtime_state(project_root, tenant_id)
    pending_auto_actions = _dedupe_string_list(runtime_state.get("pending_auto_actions", []))
    if not pending_auto_actions:
        return runtime_state, [], [], []
    executed_at = datetime.now(timezone.utc).isoformat()
    controls = _incident_controls(runtime_state)
    applied_steps: list[str] = []
    alert_messages: list[str] = []
    unsupported_steps: list[str] = []
    for step in pending_auto_actions:
        applied, alert_text = _apply_auto_action_step(controls, step, executed_at=executed_at)
        if not applied:
            unsupported_steps.append(step)
            continue
        applied_steps.append(step)
        if alert_text:
            alert_messages.append(alert_text)
    runtime_state["incident_controls"] = controls
    runtime_state["pending_auto_actions"] = []
    runtime_state["last_auto_action_execution"] = {
        "executed_at": executed_at,
        "applied_steps": applied_steps,
        "alerts": alert_messages,
        "unsupported_steps": unsupported_steps,
    }
    _save_runtime_state(project_root, tenant_id, runtime_state)
    return runtime_state, applied_steps, alert_messages, unsupported_steps


def _store_pending_auto_actions(project_root: Path, tenant_id: str, auto_actions: Any) -> None:
    runtime_state = load_runtime_state(project_root, tenant_id)
    runtime_state["pending_auto_actions"] = _dedupe_string_list(auto_actions)
    runtime_state["last_auto_actions_emitted_at"] = datetime.now(timezone.utc).isoformat()
    _save_runtime_state(project_root, tenant_id, runtime_state)


def _load_data_quality_rules(project_root: Path) -> Dict[str, Any]:
    return _read_json_dict(project_root / "config" / "data_quality_rules.json")


def _resolve_market_data_delay_gate_ms(project_root: Path, runtime_state: Dict[str, Any]) -> float:
    controls = _incident_controls(runtime_state)
    for candidate in (
        controls.get("market_data_delay_gate_ms"),
        runtime_state.get("market_data_delay_gate_ms"),
    ):
        try:
            value = float(candidate)
            if value > 0:
                return value
        except Exception:
            continue
    rules = _load_data_quality_rules(project_root)
    for check in rules.get("pre_trade_checks", []):
        if not isinstance(check, dict):
            continue
        if str(check.get("id", "") or "").strip() == "freshness_check" or str(check.get("metric", "") or "").strip() == "market_data_delay_ms":
            try:
                threshold = float(check.get("threshold", 0) or 0)
                if threshold > 0:
                    return threshold
            except Exception:
                continue
    return 2000.0


def _is_live_source_retryable_error(exc: Exception) -> bool:
    text = str(exc or "").strip()
    if not text:
        return False
    if text in LIVE_SOURCE_RETRY_ERRORS:
        return True
    return any(marker in text for marker in LIVE_SOURCE_RETRY_ERRORS)


def _acquire_pipeline_singleton_lock(project_root: Path, runtime_intent: str, source: str, execution_mode: str):
    global _PIPELINE_LOCK_HANDLE
    if runtime_intent != "live_trade":
        return None
    lock_dir = project_root / "logs"
    lock_dir.mkdir(parents=True, exist_ok=True)
    lock_path = lock_dir / f"pipeline.{runtime_intent}.{source}.{execution_mode}.lock"
    handle = lock_path.open("a+", encoding="utf-8")
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        handle.seek(0)
        owner = handle.read().strip()
        handle.close()
        owner_text = owner or "unknown_owner"
        raise SystemExit(f"pipeline singleton lock already held: {owner_text}")
    handle.seek(0)
    handle.truncate()
    handle.write(f"pid={os.getpid()} runtime_intent={runtime_intent} source={source} execution_mode={execution_mode}\n")
    handle.flush()
    _PIPELINE_LOCK_HANDLE = handle
    return handle


def _release_pipeline_singleton_lock() -> None:
    global _PIPELINE_LOCK_HANDLE
    handle = _PIPELINE_LOCK_HANDLE
    _PIPELINE_LOCK_HANDLE = None
    if handle is None:
        return
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
    except Exception:
        pass
    try:
        handle.close()
    except Exception:
        pass


@contextmanager
def _scheduler_snapshot_lock(project_root: Path):
    lock_path = project_root / "config" / "strategy_market_state.lock"
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    handle = lock_path.open("a+", encoding="utf-8")
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_SH)
        yield
    finally:
        try:
            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        except Exception:
            pass
        handle.close()


def _validate_runtime_intent(source: str, execution_mode: str, runtime_intent: str) -> None:
    src = source.lower()
    exe = execution_mode.lower()
    intent = runtime_intent.lower()
    if intent == "offline_train":
        if src != "file" or exe not in {"file", "mock"}:
            raise RuntimeError("offline_train requires source=file and execution-mode in [file,mock]")
        return
    if intent == "paper_trade":
        if src not in {"mt5", "ea_feed"} or exe not in {"mock"}:
            raise RuntimeError("paper_trade requires source in [mt5,ea_feed] and execution-mode=mock")
        return
    if intent == "live_trade":
        if src not in {"mt5", "ea_feed"} or exe not in {"direct", "ea_bridge"}:
            raise RuntimeError("live_trade requires source in [mt5,ea_feed] and execution-mode in [direct,ea_bridge]")
        return
    raise RuntimeError("runtime-intent must be one of [offline_train,paper_trade,live_trade]")


def _build_runtime_metrics(engine_request: Dict[str, Any], signal: Dict[str, Any]) -> Dict[str, Any]:
    market = engine_request.get("market_metrics", {})
    return {
        "market_data_delay_ms": float(market.get("market_data_delay_ms", 0)),
        "ack_timeout_count_5m": float(engine_request.get("runtime_feedback", {}).get("ack_timeout_count_5m", 0)),
        "slippage_points_p95_15m": float(engine_request.get("runtime_feedback", {}).get("slippage_points_p95_15m", 0)),
        "daily_token_usage_pct": float(engine_request.get("runtime_feedback", {}).get("daily_token_usage_pct", 0)),
        "governor_status": signal.get("governor_status", "APPROVED"),
    }


def _ts_to_epoch(ts: str) -> float:
    if not ts:
        return 0.0
    try:
        return datetime.fromisoformat(ts.replace("Z", "+00:00")).timestamp()
    except Exception:
        return 0.0


def _state_retry_meta(entry: Dict[str, Any]) -> Dict[str, Any]:
    history = entry.get("history", [])
    timeout_count = 0
    last_state = str(entry.get("state", "")).upper()
    last_ts = ""
    if history:
        last = history[-1]
        last_state = str(last.get("state", last_state)).upper()
        last_ts = str(last.get("ts", ""))
        for ev in history:
            if str(ev.get("state", "")).upper() == "TIMEOUT":
                timeout_count += 1
    return {"state": last_state, "last_ts": last_ts, "timeout_count": timeout_count}


def _write_json(path: Path, payload: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def _resolve_runtime_root(code_root: Path) -> Path:
    app_dir = os.environ.get(APP_DIR_ENV, "").strip()
    if app_dir:
        try:
            root = Path(app_dir).expanduser()
            root.mkdir(parents=True, exist_ok=True)
            return root
        except Exception:
            pass
    return code_root


def _resolve_runtime_path(runtime_root: Path, raw_path: str) -> Path:
    path = Path(raw_path)
    if path.is_absolute():
        return path
    return runtime_root / path


def _append_log(path: Path, payload: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    line = json.dumps(payload, ensure_ascii=False)
    with path.open("a", encoding="utf-8") as f:
        f.write(line + "\n")


def _runtime_payload(runtime_intent: str, source: str, execution_mode: str) -> Dict[str, Any]:
    return {
        "intent": runtime_intent,
        "source": source,
        "execution_mode": execution_mode,
    }


def _build_idle_signal(
    *,
    signal_id: str,
    symbol: str,
    reject_reason: str,
    evidence: list[str],
    risk_flags: list[str],
) -> Dict[str, Any]:
    now_iso = datetime.now(timezone.utc).isoformat()
    return {
        "signal_id": signal_id,
        "timestamp": now_iso,
        "symbol": symbol,
        "action": "HOLD",
        "side": "NONE",
        "mode": "UNCERTAIN",
        "governor_status": "REJECTED",
        "reject_reason": reject_reason,
        "trace": {
            "confidence": 0,
            "evidence": evidence,
            "risk_flags": risk_flags,
        },
    }


def _build_execution_report(
    signal_id: str,
    status: str,
    reason: str,
    **extra: Any,
) -> Dict[str, Any]:
    report: Dict[str, Any] = {
        "signal_id": signal_id,
        "status": status,
        "reason": reason,
    }
    for key, value in extra.items():
        if value in (None, "", [], {}):
            continue
        report[key] = value
    return report


def _build_pipeline_envelope(ctx: PipelineRunContext, outcome: PipelineOutcome) -> Dict[str, Any]:
    envelope: Dict[str, Any] = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tenant_id": ctx.tenant_id or "default",
        "signal": ctx.signal if isinstance(ctx.signal, dict) else {},
        "incidents": outcome.incidents,
        "auto_actions": outcome.auto_actions,
        "execution_report": outcome.execution_report or {},
    }
    if outcome.include_runtime:
        envelope["_runtime"] = _runtime_payload(ctx.runtime_intent, ctx.source, ctx.execution_mode)
    # Debug fields omitted from ledger to keep entries lean.
    # Raw snapshots live in state files (agent_state.json / execution_state.json).
    if ctx.promotion_info:
        envelope["_rollout"] = ctx.promotion_info
    if outcome.extra_fields:
        envelope.update(outcome.extra_fields)
    return envelope


def _compact_runtime_snapshot(payload: Any) -> Dict[str, Any]:
    if not isinstance(payload, dict):
        return {}
    compact: Dict[str, Any] = {}
    for key, value in payload.items():
        if key == "state" or value in (None, "", [], {}):
            continue
        compact[key] = deepcopy(value)
    return compact


def _derive_runtime_risk_decision(ctx: PipelineRunContext, outcome: PipelineOutcome) -> Dict[str, Any]:
    route_snapshot = (
        ctx.scheduler_snapshot.get("route", {})
        if isinstance(ctx.scheduler_snapshot.get("route", {}), dict)
        else {}
    )
    entry_owner_instance_id = str(
        route_snapshot.get(
            "entry_owner_instance_id",
            ctx.aggregated_intent.get("selected_instance_id", ctx.scheduler_snapshot.get("selected_instance_id", "")),
        )
        or ""
    )
    entry_owner_strategy_id = str(
        route_snapshot.get(
            "entry_owner_strategy_id",
            ctx.aggregated_intent.get("selected_strategy_id", ctx.scheduler_snapshot.get("selected_strategy_id", "")),
        )
        or ""
    )
    existing = (
        ctx.aggregated_intent.get("risk_decision", {})
        if isinstance(ctx.aggregated_intent.get("risk_decision", {}), dict)
        else {}
    )
    execution_report = outcome.execution_report if isinstance(outcome.execution_report, dict) else {}
    reject_reason = str(ctx.signal.get("reject_reason", "") or "")
    report_reason = str(execution_report.get("reason", "") or "")
    reason = reject_reason or report_reason
    decision = str(existing.get("decision", "") or "").lower()
    if not decision:
        decision = "allow"
        if str(execution_report.get("status", "") or "").upper() in {"REJECTED", "TIMEOUT"}:
            decision = "block"
        elif str(ctx.signal.get("action", "HOLD") or "HOLD").upper() == "HOLD" and reject_reason:
            decision = "block"
    risk_code = str(existing.get("risk_code", "") or "")
    if decision == "block" and reason:
        risk_code = reason
    guard_snapshot = _compact_runtime_snapshot(existing.get("guard_snapshot", {}))
    if not guard_snapshot:
        guard_snapshot = _compact_runtime_snapshot(ctx.signal.get("open_guard", {}))
    if not guard_snapshot:
        guard_snapshot = _compact_runtime_snapshot(ctx.signal.get("structural_conflict_guard", {}))
    if not guard_snapshot:
        binding_direction = str(ctx.signal.get("binding_direction", "") or "")
        if binding_direction:
            guard_snapshot = {"binding_direction": binding_direction}
    risk_decision = {
        **existing,
        "decision": decision,
        "execution_owner": {
            "owner_mode": "pipeline_single_track",
            "entry_owner_instance_id": entry_owner_instance_id,
            "entry_owner_strategy_id": entry_owner_strategy_id,
        },
    }
    if risk_code:
        risk_decision["risk_code"] = risk_code
    if guard_snapshot:
        risk_decision["guard_snapshot"] = guard_snapshot
    return risk_decision


def _sync_runtime_scheduler_snapshot(ctx: PipelineRunContext, outcome: PipelineOutcome) -> None:
    if ctx.state_path is None:
        return
    route_snapshot = (
        ctx.scheduler_snapshot.get("route", {})
        if isinstance(ctx.scheduler_snapshot.get("route", {}), dict)
        else {}
    )
    risk_decision = _derive_runtime_risk_decision(ctx, outcome)
    ctx.aggregated_intent["risk_decision"] = risk_decision
    payload = {
        "mode": str(ctx.scheduler_snapshot.get("mode", "single_active") or "single_active"),
        "arbiter": str(ctx.scheduler_snapshot.get("arbiter", "allocation-weighted") or "allocation-weighted"),
        "reason": str(ctx.scheduler_snapshot.get("reason", "runtime_scheduler") or "runtime_scheduler"),
        "selected_instance_id": str(
            ctx.aggregated_intent.get("selected_instance_id", ctx.scheduler_snapshot.get("selected_instance_id", "")) or ""
        ),
        "selected_strategy_id": str(
            ctx.aggregated_intent.get("selected_strategy_id", ctx.scheduler_snapshot.get("selected_strategy_id", "")) or ""
        ),
        "entry_owner_instance_id": str(
            route_snapshot.get(
                "entry_owner_instance_id",
                ctx.aggregated_intent.get("selected_instance_id", ctx.scheduler_snapshot.get("selected_instance_id", "")),
            )
            or ""
        ),
        "entry_owner_strategy_id": str(
            route_snapshot.get(
                "entry_owner_strategy_id",
                ctx.aggregated_intent.get("selected_strategy_id", ctx.scheduler_snapshot.get("selected_strategy_id", "")),
            )
            or ""
        ),
        "route": route_snapshot,
        "execution_owner": deepcopy(risk_decision.get("execution_owner", {})),
        "risk_decision": risk_decision,
        "source_versions": {
            "effective_params": str(ctx.scheduler_snapshot.get("config_revision", "manual") or "manual"),
        },
    }
    _write_json(ctx.project_root / "logs/runtime_scheduler_snapshot.json", payload)


def _emit_pipeline_outcome(ctx: PipelineRunContext, outcome: PipelineOutcome) -> Dict[str, Any]:
    _sync_runtime_scheduler_snapshot(ctx, outcome)
    envelope = _build_pipeline_envelope(ctx, outcome)
    _store_pending_auto_actions(ctx.project_root, ctx.tenant_id, outcome.auto_actions)
    _write_json(ctx.output_path, envelope)
    if outcome.sync_mailbox:
        _sync_swarm_mailbox(
            ctx.project_root,
            ctx.mt5_snapshot,
            ctx.signal,
            outcome.execution_report or None,
            ctx.promotion_info,
        )
    if outcome.append_runtime_log:
        _append_log(ctx.project_root / "logs/runtime.log", envelope)
    decision_ledger_path = ctx.project_root / "logs/decision_ledger.ndjson"
    for event_name in outcome.ledger_events:
        append_ledger_event(decision_ledger_path, event_name, envelope)
    return envelope


def _get_runtime_blackboard(project_root: Path):
    global _RUNTIME_BLACKBOARD
    target_path = str(project_root / "blackboard_state.json")
    current_path = getattr(_RUNTIME_BLACKBOARD, "file_path", None) if _RUNTIME_BLACKBOARD is not None else None
    if _RUNTIME_BLACKBOARD is None or str(current_path or "") != target_path:
        try:
            from src.swarm.blackboard import Blackboard
        except ModuleNotFoundError:
            from swarm.blackboard import Blackboard
        _RUNTIME_BLACKBOARD = Blackboard(target_path)
    return _RUNTIME_BLACKBOARD


def _runtime_blackboard_log(project_root: Path, role: str, msg_type: str, text: str) -> None:
    _get_runtime_blackboard(project_root).add_log(role, msg_type, text)


def _runtime_blackboard_update(project_root: Path, namespace: str, key: str, value: Any) -> None:
    try:
        _get_runtime_blackboard(project_root).update(namespace, key, value)
    except Exception:
        pass


def _sync_multi_strategy_blackboard(
    project_root: Path,
    scheduler_snapshot: Dict[str, Any],
    strategy_signals: list[Dict[str, Any]],
    aggregated_intent: Dict[str, Any],
    report: Optional[Dict[str, Any]] = None,
) -> None:
    contributor_rows = (
        aggregated_intent.get("contributors", [])
        if isinstance(aggregated_intent.get("contributors", []), list)
        else []
    )
    selected_instance_id = str(aggregated_intent.get("selected_instance_id", "") or "")
    selected_strategy_id = str(aggregated_intent.get("selected_strategy_id", "") or "")
    instance_rows: Dict[str, Dict[str, Any]] = {}
    aggregated_action = str(aggregated_intent.get("action", "HOLD") or "HOLD").upper()
    aggregated_side = str(aggregated_intent.get("side", "NONE") or "NONE").upper()
    blocked_reasons = [
        str(item) for item in aggregated_intent.get("blocked_reasons", [])
        if str(item).strip()
    ] if isinstance(aggregated_intent.get("blocked_reasons", []), list) else []
    selected_reason = str(aggregated_intent.get("selected_reason", "") or "")
    non_risk_hold_reasons = {
        "AGGREGATED_HOLD_LOW_NET_SCORE",
        "AGGREGATED_HOLD_LOW_ACTIONABLE_WEIGHT",
        "AGGREGATED_HOLD_NO_ACTIONABLE_SIGNAL",
    }
    governor_blockers = [
        reason
        for reason in blocked_reasons
        if not any(marker in reason for marker in non_risk_hold_reasons)
    ]
    for signal in strategy_signals:
        if not isinstance(signal, dict):
            continue
        instance_id = str(signal.get("_instance_id", "") or "").strip()
        if not instance_id:
            continue
        side = str(signal.get("side", "NONE") or "NONE").upper()
        action = str(signal.get("action", "HOLD") or "HOLD").upper()
        signed_score = _signal_signed_score(signal, float(signal.get("instance_weight", 0.0) or 0.0))
        suppression_reason = ""
        if aggregated_action == "HOLD" and action != "HOLD":
            suppression_reason = "conflict_or_low_net_score"
        elif action != "HOLD" and side != aggregated_side:
            suppression_reason = "opposite_side_offset"
        instance_rows[instance_id] = {
            "strategy_id": str(signal.get("_sig_strategy_id", signal.get("strategy_id", "")) or ""),
            "action": action,
            "side": side,
            "confidence": round(float((signal.get("trace", {}) or {}).get("confidence", 0.0) or 0.0), 4),
            "signed_score": round(float(signed_score), 6),
            "weight": round(float(signal.get("instance_weight", 0.0) or 0.0), 4),
            "selected": instance_id == selected_instance_id,
            "suppressed": bool(suppression_reason),
            "suppression_reason": suppression_reason,
            "raw_reason": _signal_reason_text(signal),
            "reject_reason": str(signal.get("reject_reason", "") or ""),
        }
    # ── Per-instance health scoring (computed once, used for both risk_state and health_score) ──
    hub = get_intelligence_hub()
    instances_for_health = scheduler_snapshot.get("instances", []) if isinstance(scheduler_snapshot.get("instances", []), list) else []
    health_results = hub.refresh_all(project_root, instances_for_health)
    instance_health_score = {
        iid: h.score
        for iid, h in health_results.items()
    }
    for iid, h in health_results.items():
        if iid not in instance_health_score:
            instance_health_score[iid] = 50.0
    for instance in instances_for_health:
        iid = str(instance.get("instance_id", "") or "").strip()
        if iid and iid not in instance_health_score:
            instance_health_score[iid] = 50.0

    instance_risk_state: Dict[str, str] = {}
    for row in contributor_rows:
        if not isinstance(row, dict):
            continue
        iid = str(row.get("instance_id", "") or "").strip()
        if not iid:
            continue
        base_state = "blocked" if str(row.get("reason", "") or "").strip() else "normal"
        health = health_results.get(iid)
        if health:
            if health.suggested_action == "shadow_only":
                base_state = "blocked"
            elif health.suggested_action == "rehabilitate":
                base_state = "rehab"
            elif health.suggested_action == "reduce_size" and base_state == "normal":
                base_state = "reduced"
        instance_risk_state[iid] = base_state
    for iid, health in health_results.items():
        if iid not in instance_risk_state:
            if health.suggested_action == "shadow_only":
                instance_risk_state[iid] = "blocked"
            elif health.suggested_action == "rehabilitate":
                instance_risk_state[iid] = "rehab"
            elif health.suggested_action == "reduce_size":
                instance_risk_state[iid] = "reduced"
            else:
                instance_risk_state[iid] = "normal"
    global_state = "blocked" if governor_blockers else "normal"
    if isinstance(report, dict) and str(report.get("status", "") or "").upper() in {"REJECTED", "TIMEOUT", "RETRY_EXHAUSTED", "INIT_FAILED"}:
        global_state = "blocked"

    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "mode", str(scheduler_snapshot.get("mode", "single_active") or "single_active"))
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "arbiter", str(scheduler_snapshot.get("arbiter", "allocation-weighted") or "allocation-weighted"))
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "reason", str(scheduler_snapshot.get("reason", "runtime_scheduler") or "runtime_scheduler"))
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "active_strategy_ids", list(scheduler_snapshot.get("active_strategy_ids", [])) if isinstance(scheduler_snapshot.get("active_strategy_ids", []), list) else [])
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "active_instance_ids", list(scheduler_snapshot.get("active_instance_ids", [])) if isinstance(scheduler_snapshot.get("active_instance_ids", []), list) else [])
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "strategy_weights", dict(scheduler_snapshot.get("strategy_weights", {})) if isinstance(scheduler_snapshot.get("strategy_weights", {}), dict) else {})
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "instance_weights", dict(scheduler_snapshot.get("instance_weights", {})) if isinstance(scheduler_snapshot.get("instance_weights", {}), dict) else {})
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "selected_strategy_id", selected_strategy_id)
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "selected_instance_id", selected_instance_id)
    route_snapshot = (
        scheduler_snapshot.get("route", {})
        if isinstance(scheduler_snapshot.get("route", {}), dict)
        else {}
    )
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "route", route_snapshot)
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "route_family", str(route_snapshot.get("route_family", "") or ""))
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "allowed_side", str(route_snapshot.get("allowed_side", "BOTH") or "BOTH"))
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "candidate_strategy_ids", list(route_snapshot.get("candidate_strategy_ids", [])) if isinstance(route_snapshot.get("candidate_strategy_ids", []), list) else [])
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "router_weights", dict(route_snapshot.get("weights", {})) if isinstance(route_snapshot.get("weights", {}), dict) else {})
    _runtime_blackboard_update(project_root, "Scheduler_Runtime", "direction_guard", dict(route_snapshot.get("direction_guard", {})) if isinstance(route_snapshot.get("direction_guard", {}), dict) else {})

    _runtime_blackboard_update(project_root, "Strategy_Instances", "instances", instance_rows)
    _runtime_blackboard_update(project_root, "Strategy_Instances", "selected_strategy_id", selected_strategy_id)
    _runtime_blackboard_update(project_root, "Strategy_Instances", "selected_instance_id", selected_instance_id)
    _runtime_blackboard_update(project_root, "Strategy_Instances", "contributors_count", len(contributor_rows))

    _runtime_blackboard_update(project_root, "Execution_Arbitration", "final_action", aggregated_action)
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "final_side", aggregated_side)
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "net_score", round(float(aggregated_intent.get("net_score", 0.0) or 0.0), 6))
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "confidence", round(float(aggregated_intent.get("confidence", 0.0) or 0.0), 4))
    _runtime_blackboard_update(
        project_root,
        "Execution_Arbitration",
        "hold_mode",
        "blocked_hold" if aggregated_action == "HOLD" and governor_blockers else "signal_hold" if aggregated_action == "HOLD" and blocked_reasons else "selected_action",
    )
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "blocked_reasons", blocked_reasons)
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "selected_reason", selected_reason)
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "hold_diagnostics", _hold_diagnostics_summary(aggregated_intent) if aggregated_action == "HOLD" else {})
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "selected_strategy_id", selected_strategy_id)
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "selected_instance_id", selected_instance_id)
    _runtime_blackboard_update(project_root, "Execution_Arbitration", "last_execution_status", str(report.get("status", "") or "") if isinstance(report, dict) else "")

    _runtime_blackboard_update(project_root, "Risk_Protocol", "governor_blockers", governor_blockers)
    _runtime_blackboard_update(project_root, "Risk_Protocol", "active_strategy_ids", list(scheduler_snapshot.get("active_strategy_ids", [])) if isinstance(scheduler_snapshot.get("active_strategy_ids", []), list) else [])
    _runtime_blackboard_update(project_root, "Risk_Protocol", "active_instance_ids", list(scheduler_snapshot.get("active_instance_ids", [])) if isinstance(scheduler_snapshot.get("active_instance_ids", []), list) else [])
    _runtime_blackboard_update(project_root, "Risk_Protocol", "instance_risk_state", instance_risk_state)
    _runtime_blackboard_update(project_root, "Risk_Protocol", "instance_health_score", instance_health_score)
    _runtime_blackboard_update(project_root, "Risk_Protocol", "per_instance_health_details", {
        iid: h.to_dict()
        for iid, h in health_results.items()
    })
    _runtime_blackboard_update(project_root, "Risk_Protocol", "health_alerts", list(scheduler_snapshot.get("health_alerts", [])) if isinstance(scheduler_snapshot.get("health_alerts", []), list) else [])
    _runtime_blackboard_update(project_root, "Risk_Protocol", "cross_strategy_correlation", aggregated_intent.get("correlation", {}))
    _runtime_blackboard_update(project_root, "Risk_Protocol", "voting_weights", aggregated_intent.get("voting_weights", {}))
    _runtime_blackboard_update(project_root, "Risk_Protocol", "global_state", global_state)


def _mailbox_emit(project_root: Path, role: str, msg_type: str, text: str, dedupe_key: str, min_interval_sec: float) -> None:
    now = time.time()
    last_ts = _MAILBOX_LAST_EMIT.get(dedupe_key, 0.0)
    if now - last_ts < min_interval_sec:
        return
    _MAILBOX_LAST_EMIT[dedupe_key] = now
    _runtime_blackboard_log(project_root, role, msg_type, text)


def _emit_scheduler_health_alerts(project_root: Path, scheduler_snapshot: Dict[str, Any]) -> None:
    for alert in _dedupe_string_list(scheduler_snapshot.get("health_alerts", [])):
        _mailbox_emit(
            project_root,
            "Risk_Commissar",
            "WARN",
            f"Runtime scheduler alert: {alert}",
            f"scheduler_alert:{alert}",
            60.0,
        )


def _signal_reason_text(signal: Dict[str, Any]) -> str:
    trace = signal.get("trace", {}) if isinstance(signal.get("trace", {}), dict) else {}
    evidence = trace.get("evidence", [])
    if isinstance(evidence, list) and evidence:
        return " | ".join(str(item) for item in evidence[:4])
    reject_reason = str(signal.get("reject_reason", "") or "").strip()
    if reject_reason:
        return reject_reason
    return str(signal.get("mode", "UNCERTAIN") or "UNCERTAIN")


def _hold_diagnostics_summary(aggregated_intent: Dict[str, Any]) -> dict[str, Any]:
    contributors = (
        aggregated_intent.get("contributors", [])
        if isinstance(aggregated_intent.get("contributors", []), list)
        else []
    )
    blocked_reasons = [
        str(item) for item in aggregated_intent.get("blocked_reasons", [])
        if str(item).strip()
    ] if isinstance(aggregated_intent.get("blocked_reasons", []), list) else []
    suppressed: list[dict[str, Any]] = []
    for row in contributors:
        if not isinstance(row, dict):
            continue
        action = str(row.get("action", "HOLD") or "HOLD").upper()
        reason = str(row.get("reason", "") or "").strip()
        if action == "HOLD" or reason:
            suppressed.append(
                {
                    "instance_id": str(row.get("instance_id", "") or ""),
                    "strategy_id": str(row.get("strategy_id", "") or ""),
                    "action": action,
                    "side": str(row.get("side", "NONE") or "NONE").upper(),
                    "reason": reason or "INSTANCE_SIGNAL_HOLD",
                    "confidence": float(row.get("confidence", 0.0) or 0.0),
                    "weight": float(row.get("weight", 0.0) or 0.0),
                    "signed_score": float(row.get("signed_score", 0.0) or 0.0),
                }
            )
    return {
        "blocked_reasons": blocked_reasons,
        "suppressed_instances": suppressed,
        "actionable_count": int(aggregated_intent.get("actionable_count", 0) or 0),
        "actionable_weight_total": float(aggregated_intent.get("actionable_weight_total", 0.0) or 0.0),
        "actionable_weight_floor": float(aggregated_intent.get("actionable_weight_floor", 0.0) or 0.0),
        "net_score": float(aggregated_intent.get("net_score", 0.0) or 0.0),
        "raw_net_score": float(aggregated_intent.get("raw_net_score", 0.0) or 0.0),
        "selected_instance_id": str(aggregated_intent.get("selected_instance_id", "") or ""),
        "selected_strategy_id": str(aggregated_intent.get("selected_strategy_id", "") or ""),
    }


def _format_shadow_gate_reason(rollout_info: Dict[str, Any]) -> str:
    gates = rollout_info.get("gate_readiness", []) if isinstance(rollout_info.get("gate_readiness", []), list) else []
    blocked = [gate for gate in gates if isinstance(gate, dict) and not bool(gate.get("ready", False))]
    if not blocked:
        return "尚未满足 shadow 升级门禁"
    top = []
    for gate in blocked[:3]:
        top.append(f"{gate.get('gate')}={gate.get('current')}/{gate.get('required')}")
    next_mode = str(blocked[0].get("next_mode", "paper") or "paper")
    return f"shadow->{next_mode} 未通过：{', '.join(top)}"


def _sync_swarm_mailbox(
    project_root: Path,
    mt5_snapshot: Dict[str, Any],
    signal: Dict[str, Any],
    report: Optional[Dict[str, Any]],
    rollout_info: Optional[Dict[str, Any]] = None,
) -> None:
    market = mt5_snapshot.get("market", {}) if isinstance(mt5_snapshot.get("market", {}), dict) else {}
    bid = float(
        mt5_snapshot.get("bid", mt5_snapshot.get("Bid", market.get("bid", market.get("Bid", 0.0)))) or 0.0
    )
    ask = float(
        mt5_snapshot.get("ask", mt5_snapshot.get("Ask", market.get("ask", market.get("Ask", 0.0)))) or 0.0
    )
    mode = str(signal.get("mode", "UNCERTAIN") or "UNCERTAIN")
    if bid > 0 and ask > 0:
        quote_text = f"探明最新报价 {bid:.2f}/{ask:.2f}，当前环境: {mode}"
        quote_key = f"quote:{mode}:{round(bid, 2)}:{round(ask, 2)}"
        _mailbox_emit(project_root, "Micro_Sniper", "REPORT", quote_text, quote_key, 30.0)

    action = str(signal.get("action", "HOLD") or "HOLD").upper()
    side = str(signal.get("side", "NONE") or "NONE").upper()
    confidence = float((signal.get("trace", {}) or {}).get("confidence", 0.0) or 0.0) * 100.0
    reason_text = _signal_reason_text(signal)
    inferred_side = "SELL" if "lt" in reason_text.lower() or "bear" in reason_text.lower() else "BUY"

    if action != "HOLD" and side in {"BUY", "SELL"}:
        side_zh = "做多(BUY)" if side == "BUY" else "做空(SELL)"
        alert_text = f"报告政委：我捕捉到了一个强烈的 {side_zh} 信号（置信度: {confidence:.1f}%）。理由是 {reason_text}。请求战术指导。"
        _mailbox_emit(project_root, "Micro_Sniper", "alert", alert_text, f"signal:{signal.get('signal_id')}", 1.0)
    elif confidence >= 60.0:
        side_zh = "做空(SELL)" if inferred_side == "SELL" else "做多(BUY)"
        alert_text = f"发现潜在 {side_zh} 机会（置信度: {confidence:.1f}%），但当前执行结果仍为 HOLD。线索：{reason_text}。"
        _mailbox_emit(project_root, "Micro_Sniper", "alert", alert_text, f"opportunity:{signal.get('signal_id')}", 2.0)

    if report:
        status = str(report.get("status", "") or "").upper()
        reason = str(report.get("reason", "") or "").strip()
        if status == "REJECTED" and reason == "SHADOW_MODE_BLOCK_REAL_EXECUTION":
            risk_text = f"检测到历史 shadow 拦单结果。当前策略已改为 shadow 仅作离线观察，生产执行不再被其阻断。旧门禁信息：{_format_shadow_gate_reason(rollout_info or {})}。"
            _mailbox_emit(project_root, "Risk_Commissar", "INFO", risk_text, f"risk:shadow:{signal.get('signal_id')}", 20.0)
        elif status == "REJECTED":
            risk_text = f"本轮请求已驳回。原因：{reason or 'UNKNOWN_REJECT_REASON'}。"
            _mailbox_emit(project_root, "Risk_Commissar", "WARN", risk_text, f"risk:reject:{signal.get('signal_id')}:{reason}", 20.0)
        elif status in {"ACKED", "FILLED"} and side in {"BUY", "SELL"}:
            side_zh = "做多" if side == "BUY" else "做空"
            risk_text = f"已批准 {side_zh} 请求，执行状态：{status}。"
            _mailbox_emit(project_root, "Risk_Commissar", "success", risk_text, f"risk:exec:{signal.get('signal_id')}:{status}", 20.0)
    elif action == "HOLD" and confidence >= 60.0:
        side_zh = "做空" if inferred_side == "SELL" else "做多"
        risk_text = (
            f"已收到 {side_zh} 线索，但当前未达到出击条件。"
            f"依据：{reason_text}。继续潜伏。"
        )
        _mailbox_emit(project_root, "Risk_Commissar", "INFO", risk_text, f"risk:hold:{signal.get('signal_id')}", 20.0)


def _load_snapshot(source: str, input_path: Path, live_client: Optional[Any]) -> Dict[str, Any]:
    if source in {"mt5", "ea_feed"}:
        if live_client is None:
            raise RuntimeError("LIVE_SOURCE_NOT_CONNECTED")
        snap = live_client.snapshot()
        if snap is None:
            raise RuntimeError("LIVE_SOURCE_SNAPSHOT_UNAVAILABLE")
        return snap
    return load_mt5_snapshot(input_path)


def _load_effective_payload(project_root: Path) -> Dict[str, Any]:
    path = project_root / "logs" / "effective_params.json"
    if not path.exists():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        return payload if isinstance(payload, dict) else {}
    except Exception:
        return {}


def _load_runtime_instances(project_root: Path) -> list[dict[str, Any]]:
    path = project_root / "config" / "strategy_market_state.json"
    if not path.exists():
        return []
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return []
    runtime_instances = payload.get("runtime_instances", []) if isinstance(payload.get("runtime_instances", []), list) else []
    return [item for item in runtime_instances if isinstance(item, dict)]


def _load_scheduler_snapshot(project_root: Path, engine_request: Dict[str, Any]) -> Dict[str, Any]:
    with _scheduler_snapshot_lock(project_root):
        effective_payload = _load_effective_payload(project_root)
        scheduler = (
            effective_payload.get("runtime_scheduler", {})
            if isinstance(effective_payload.get("runtime_scheduler", {}), dict)
            else {}
        )
        runtime_instances = _load_runtime_instances(project_root)
    instance_map = {
        str(item.get("instance_id", "")).strip(): item
        for item in runtime_instances
        if str(item.get("instance_id", "")).strip()
    }
    # When server-provided runtime_scheduler is empty, read local active_instances.json
    scheduler_has_active = bool(
        scheduler.get("active_instance_ids")
        or scheduler.get("active_instances")
        or scheduler.get("active_strategy_ids")
    )
    if not scheduler_has_active:
        local_config = _read_json_dict(project_root / "config" / "active_instances.json")
        if local_config:
            local_instances = local_config.get("instances", [])
            if isinstance(local_instances, list) and local_instances:
                scheduler = {
                    "mode": str(local_config.get("mode", "single_active") or "single_active"),
                    "arbiter": str(local_config.get("arbiter", "allocation-weighted") or "allocation-weighted"),
                    "reason": "local_active_instances",
                    "active_instance_ids": [],
                    "active_strategy_ids": [],
                    "instance_weights": {},
                    "strategy_weights": {},
                    "strategy_instance_map": {},
                }
                for inst in local_instances:
                    if not isinstance(inst, dict):
                        continue
                    if not inst.get("enabled", True):
                        continue
                    iid = str(inst.get("id", "")).strip()
                    sid = str(inst.get("strategy_id", "")).strip()
                    if not iid or not sid:
                        continue
                    scheduler["active_instance_ids"].append(iid)
                    if sid not in scheduler["active_strategy_ids"]:
                        scheduler["active_strategy_ids"].append(sid)
                    scheduler["instance_weights"][iid] = 1.0
                    scheduler["strategy_weights"][sid] = scheduler["strategy_weights"].get(sid, 0.0) + 1.0
                    scheduler["strategy_instance_map"].setdefault(sid, []).append(iid)
                    if iid not in instance_map:
                        instance_map[iid] = {
                            "instance_id": iid,
                            "strategy_id": sid,
                            "strategy_name": sid,
                            "role": "live_primary",
                            "allocation": 1.0,
                            "param_overrides": inst.get("params", {}),
                        }
    raw_active_instance_values = scheduler.get("active_instance_ids", scheduler.get("active_instances", []))
    active_instance_source_present = "active_instance_ids" in scheduler or "active_instances" in scheduler
    active_instance_ids = [
        str(item)
        for item in raw_active_instance_values
        if str(item).strip()
    ] if isinstance(raw_active_instance_values, list) else []
    active_strategy_ids = [
        str(item)
        for item in scheduler.get("active_strategy_ids", [])
        if str(item).strip()
    ] if isinstance(scheduler.get("active_strategy_ids", []), list) else []
    instance_weights = (
        scheduler.get("instance_weights", {})
        if isinstance(scheduler.get("instance_weights", {}), dict)
        else {}
    )
    strategy_weights = (
        scheduler.get("strategy_weights", {})
        if isinstance(scheduler.get("strategy_weights", {}), dict)
        else {}
    )
    strategy_instance_map = (
        scheduler.get("strategy_instance_map", {})
        if isinstance(scheduler.get("strategy_instance_map", {}), dict)
        else {}
    )
    config_revision = str(
        scheduler.get("config_revision")
        or effective_payload.get("config_revision")
        or effective_payload.get("revision")
        or "manual"
    )
    health_alerts: list[str] = []
    if active_instance_source_present and not active_instance_ids:
        health_alerts.append("ACTIVE_INSTANCE_IDS_EMPTY")
    instances: list[dict[str, Any]] = []
    for instance_id in active_instance_ids:
        row = instance_map.get(instance_id, {})
        if not row:
            continue
        instances.append(
            {
                "instance_id": instance_id,
                "strategy_id": str(row.get("strategy_id", "")).strip(),
                "strategy_name": str(row.get("strategy_name", row.get("strategy_id", "")) or ""),
                "role": str(row.get("role", "live_support") or "live_support"),
                "allocation": float(row.get("allocation", 0.0) or 0.0),
                "weight": float(instance_weights.get(instance_id, 0.0) or 0.0),
                "config_revision": str(row.get("config_revision", config_revision) or config_revision),
                "param_overrides": row.get("param_overrides", {}) if isinstance(row.get("param_overrides", {}), dict) else {},
            }
        )
    if active_instance_ids and not instances:
        health_alerts.append("ACTIVE_INSTANCE_IDS_UNRESOLVED")
    if not instances:
        fallback_strategy = str(
            effective_payload.get("strategy_type")
            or engine_request.get("ai_trade_plan", {}).get("strategy_id")
            or "ema_macd"
        )
        fallback_instance_id = f"legacy:{fallback_strategy}"
        instances = [
            {
                "instance_id": fallback_instance_id,
                "strategy_id": fallback_strategy,
                "strategy_name": fallback_strategy,
                "role": "live_primary",
                "allocation": 1.0,
                "weight": 1.0,
                "config_revision": config_revision,
                "param_overrides": {},
            }
        ]
        active_instance_ids = [fallback_instance_id]
        active_strategy_ids = [fallback_strategy]
        instance_weights = {fallback_instance_id: 1.0}
        strategy_weights = {fallback_strategy: 1.0}
        strategy_instance_map = {fallback_strategy: [fallback_instance_id]}
        scheduler = {
            "mode": "single_active",
            "arbiter": "legacy-single-strategy",
            "reason": "legacy_single_strategy_fallback",
        }
        if active_instance_source_present:
            health_alerts.append("ACTIVE_INSTANCE_IDS_FALLBACK_TO_LEGACY")
    return {
        "mode": str(scheduler.get("mode", "single_active") or "single_active"),
        "arbiter": str(scheduler.get("arbiter", "allocation-weighted") or "allocation-weighted"),
        "reason": str(scheduler.get("reason", "runtime_scheduler") or "runtime_scheduler"),
        "config_revision": config_revision,
        "active_instance_ids": active_instance_ids,
        "active_strategy_ids": active_strategy_ids,
        "instance_weights": {str(key): float(value or 0.0) for key, value in instance_weights.items()},
        "strategy_weights": {str(key): float(value or 0.0) for key, value in strategy_weights.items()},
        "strategy_instance_map": {
            str(key): [str(item) for item in value if str(item).strip()]
            for key, value in strategy_instance_map.items()
            if isinstance(value, list)
        },
        "instances": instances,
        "health_alerts": health_alerts,
        "decision_policy": (
            scheduler.get("decision_policy", {})
            if isinstance(scheduler.get("decision_policy", {}), dict)
            else {}
        ),
    }


def _safe_float(value: Any, default: float = 0.0) -> float:
    try:
        return float(value)
    except Exception:
        return default


def _normalize_regime_label(value: Any) -> str:
    return str(value or "").strip().upper().replace("-", "_").replace(" ", "_")


def _coerce_bool(value: Any, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in {"1", "true", "yes", "on"}:
            return True
        if lowered in {"0", "false", "no", "off"}:
            return False
    return default


def _bounded_float(value: Any, default: float, *, low: float = 0.0, high: float = 2.0) -> float:
    try:
        return round(max(low, min(high, float(value))), 4)
    except Exception:
        return default


def _normalize_string_list(values: Any) -> list[str]:
    if not isinstance(values, list):
        return []
    normalized: list[str] = []
    for item in values:
        text = str(item or "").strip()
        if text and text not in normalized:
            normalized.append(text)
    return normalized


def _normalize_entry_style(value: Any, default: str = "MARKET") -> str:
    text = str(value or default).strip().upper().replace("-", "_").replace(" ", "_")
    style_aliases = {
        "MARKET": "MARKET",
        "LIMIT": "LIMIT",
        "PULLBACK_LIMIT": "PULLBACK_LIMIT",
        "CONFIRMATION_ONLY": "CONFIRMATION_ONLY",
    }
    return style_aliases.get(text, default)


def _normalize_entry_style_list(values: Any, fallback: list[str]) -> list[str]:
    items = values if isinstance(values, list) else fallback
    normalized: list[str] = []
    for item in items:
        style = _normalize_entry_style(item)
        if style not in normalized:
            normalized.append(style)
    return normalized or list(fallback)


def _default_execution_advice_matrix() -> dict[str, dict[str, Any]]:
    return {
        "ALLOW_FOLLOW": {
            "allowed_entry_styles": ["MARKET", "LIMIT"],
            "confirmation_required": False,
            "probe_size_scale": 0.6,
        },
        "LIMIT_ONLY": {
            "allowed_entry_styles": ["LIMIT", "PULLBACK_LIMIT"],
            "confirmation_required": False,
            "probe_size_scale": 0.45,
        },
        "WAIT_CONFIRMATION": {
            "allowed_entry_styles": ["LIMIT", "PULLBACK_LIMIT", "CONFIRMATION_ONLY"],
            "confirmation_required": True,
            "probe_size_scale": 0.35,
        },
        "BLOCK_COUNTER_TREND": {
            "allowed_entry_styles": ["MARKET", "LIMIT"],
            "confirmation_required": False,
            "probe_size_scale": 0.25,
        },
    }


def _normalize_execution_advice_matrix(raw: Any) -> dict[str, dict[str, Any]]:
    base = _default_execution_advice_matrix()
    payload = raw if isinstance(raw, dict) else {}
    normalized: dict[str, dict[str, Any]] = {}
    for advice, defaults in base.items():
        row = payload.get(advice, {}) if isinstance(payload.get(advice, {}), dict) else {}
        normalized[advice] = {
            "allowed_entry_styles": _normalize_entry_style_list(
                row.get("allowed_entry_styles", defaults["allowed_entry_styles"]),
                defaults["allowed_entry_styles"],
            ),
            "confirmation_required": _coerce_bool(
                row.get("confirmation_required", defaults["confirmation_required"]),
                defaults["confirmation_required"],
            ),
            "probe_size_scale": _bounded_float(
                row.get("probe_size_scale", defaults["probe_size_scale"]),
                defaults["probe_size_scale"],
                low=0.1,
                high=1.0,
            ),
        }
    return normalized


def _normalize_router_family(
    raw: Any,
    *,
    primary_strategy_id: str,
    support_strategy_ids: list[str],
    rank_weights: list[float],
) -> dict[str, Any]:
    payload = raw if isinstance(raw, dict) else {}
    primary = str(payload.get("primary_strategy_id", primary_strategy_id) or primary_strategy_id).strip()
    supports = _normalize_string_list(payload.get("support_strategy_ids", support_strategy_ids))
    candidates = _normalize_string_list([primary, *supports])
    if not candidates:
        candidates = _normalize_string_list([primary_strategy_id, *support_strategy_ids])
    raw_weights = payload.get("rank_weights", rank_weights)
    rank_values = raw_weights if isinstance(raw_weights, list) else rank_weights
    normalized_weights: list[float] = []
    for idx in range(len(candidates)):
        default_weight = max(0.0, float(rank_weights[idx] if idx < len(rank_weights) else 0.0))
        candidate_weight = rank_values[idx] if idx < len(rank_values) else default_weight
        normalized_weights.append(
            _bounded_float(candidate_weight, default_weight, low=0.0, high=1.0)
        )
    total = sum(normalized_weights)
    if total <= 0:
        normalized_weights = [max(0.0, float(item or 0.0)) for item in rank_weights[: len(candidates)]]
        total = sum(normalized_weights)
    if total <= 0:
        normalized_weights = [round(1.0 / max(1, len(candidates)), 4)] * len(candidates)
    else:
        normalized_weights = [round(weight / total, 4) for weight in normalized_weights]
    return {
        "primary_strategy_id": candidates[0],
        "support_strategy_ids": candidates[1:],
        "rank_weights": normalized_weights,
    }


def _resolve_rank_weights(rank_weights: list[float], count: int) -> list[float]:
    if count <= 0:
        return []
    weights = [max(0.0, float(item or 0.0)) for item in rank_weights[:count]]
    if len(weights) < count:
        weights.extend([0.0] * (count - len(weights)))
    total = sum(weights)
    if total <= 0:
        return [round(1.0 / max(1, count), 4)] * count
    return [round(weight / total, 6) for weight in weights]


def _load_recent_closures(project_root: Path) -> list[Dict[str, Any]]:
    state_path = project_root / "data" / "bridge" / "state" / "agent_state.json"
    try:
        payload = json.loads(state_path.read_text(encoding="utf-8"))
    except Exception:
        return []
    rows = payload.get("recent_closures", []) if isinstance(payload, dict) else []
    if not isinstance(rows, list):
        return []
    return [row for row in rows if isinstance(row, dict)]


def _loss_streak(rows: list[Dict[str, Any]]) -> int:
    streak = 0
    for row in reversed(rows):
        try:
            pnl_value = float(row.get("pnl", 0.0) or 0.0)
        except Exception:
            continue
        if pnl_value < 0:
            streak += 1
        else:
            break
    return streak


def _hit_rate(rows: list[Dict[str, Any]]) -> float:
    if not rows:
        return 0.0
    hits = 0
    for row in rows:
        try:
            if float(row.get("pnl", 0.0) or 0.0) > 0:
                hits += 1
        except Exception:
            continue
    return hits / max(1, len(rows))


def _inject_account_risk_metrics(engine_request: Dict[str, Any], project_root: Path) -> None:
    """用 agent_state.json 里的真实数据覆盖 account_metrics 中的硬编码 0。

    ea_bridge_live / mt5_live 的 snapshot 将 consecutive_losses 和
    daily_realized_loss_pct 硬编码为 0，导致 loss_streak_cooldown 和
    daily_loss_circuit_breaker 熔断规则永远不触发。此函数从 state 文件
    读取实际平仓记录，计算真实值并注入 engine_request。
    """
    rows = _load_recent_closures(project_root)
    account = engine_request.get("account_metrics", {})
    if not isinstance(account, dict):
        account = {}
        engine_request["account_metrics"] = account

    # ── consecutive_losses：从最近平仓倒序扫描连亏 ──
    streak = _loss_streak(rows)
    account["consecutive_losses"] = int(streak)

    # ── daily_realized_loss_pct：今日累计已实现亏损占权益百分比 ──
    if rows:
        try:
            today_start = int(datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0).timestamp())
            today_pnl = sum(
                float(row.get("pnl", 0.0) or 0.0)
                for row in rows
                if isinstance(row, dict) and int(row.get("ts", 0) or 0) >= today_start
            )
            equity = float(account.get("equity", 0) or 1.0)
            if today_pnl < 0 and equity > 0:
                account["daily_realized_loss_pct"] = round(abs(today_pnl) / max(1.0, equity) * 100.0, 4)
            else:
                account["daily_realized_loss_pct"] = 0.0
        except Exception:
            account.setdefault("daily_realized_loss_pct", 0.0)
    else:
        account.setdefault("daily_realized_loss_pct", 0.0)


def _sync_agent_state_from_mt5(ctx: PipelineRunContext) -> None:
    """每轮将 MT5 真实持仓同步到 agent_state.json，消除幽灵单。

    gold_trader.py 已不在主循环中运行（launcher 强制切到 pipeline 单入口），
    但 GUI 仍从 agent_state.json 读取持仓展示。pipeline 每轮 call 此函数，
    确保 agent_state.json 的 position_open / positions_all 与 MT5 保持同步。
    """
    if ctx.source not in {"mt5", "ea_feed"}:
        return
    snap = getattr(ctx, "mt5_snapshot", None)
    if not isinstance(snap, dict):
        return
    state_path = ctx.project_root / "data" / "bridge" / "state" / "agent_state.json"
    try:
        base = json.loads(state_path.read_text(encoding="utf-8")) if state_path.exists() else {}
    except Exception:
        base = {}
    if not isinstance(base, dict):
        base = {}

    positions_all = snap.get("positions_all", [])
    position_open = bool(positions_all)

    base["position_open"] = position_open
    base["positions_all"] = positions_all
    account = snap.get("account", {}) if isinstance(snap.get("account", {}), dict) else {}
    equity = float(account.get("equity", 0.0) or 0.0)
    if equity > 0:
        base["balance"] = equity
        base["equity"] = equity
    elif not base.get("balance"):
        pass  # preserve existing balance/equity if already set

    if position_open and positions_all:
        p = positions_all[0] if isinstance(positions_all[0], dict) else {}
        base["position_ticket"] = int(p.get("ticket", 0) or 0)
        base["position_side"] = str(p.get("side", "BUY") or "BUY").upper()
        base["position_entry"] = float(p.get("entry", 0.0) or 0.0)
        base["position_price"] = float(p.get("entry", 0.0) or 0.0)
        base["position_sl"] = float(p.get("sl", 0.0) or 0.0)
        base["position_tp"] = float(p.get("tp", 0.0) or 0.0)
        base["position_vol"] = float(p.get("volume", 0.0) or 0.0)
    else:
        base["position_ticket"] = 0
        base["position_side"] = None
        base["position_entry"] = 0.0
        base["position_price"] = 0.0
        base["position_sl"] = 0.0
        base["position_tp"] = 0.0
        base["position_vol"] = 0.0

    try:
        state_path.parent.mkdir(parents=True, exist_ok=True)
        tmp = state_path.with_suffix(state_path.suffix + ".tmp")
        tmp.write_text(json.dumps(base, ensure_ascii=False, default=str), encoding="utf-8")
        os.replace(tmp, state_path)
    except Exception:
        pass


def _compute_direction_loss_guard(
    project_root: Path,
    macro_state: Dict[str, Any],
    policy: Dict[str, Any],
) -> Dict[str, Any]:
    macro_policy = policy.get("macro", {}) if isinstance(policy.get("macro", {}), dict) else {}
    guard_policy = (
        macro_policy.get("direction_loss_guard", {})
        if isinstance(macro_policy.get("direction_loss_guard", {}), dict)
        else {}
    )
    result = {
        "enabled": bool(guard_policy.get("enabled", True)),
        "state": "inactive",
        "reason": "",
        "scale": 1.0,
        "sample_count": 0,
        "aligned_count": 0,
        "counter_count": 0,
        "aligned_hit_rate": 0.0,
        "counter_hit_rate": 0.0,
        "aligned_loss_streak": 0,
        "effective_direction": str(macro_state.get("effective_direction", "NEUTRAL") or "NEUTRAL"),
        "disempowered": False,
    }
    macro_direction = str(macro_state.get("direction", "NEUTRAL") or "NEUTRAL").upper()
    if not result["enabled"] or macro_direction not in {"LONG", "SHORT"}:
        return result

    rows = _load_recent_closures(project_root)
    lookback = int(float(guard_policy.get("lookback", 8) or 8))
    min_samples = int(float(guard_policy.get("min_samples", 3) or 3))
    relevant_rows = [
        row
        for row in rows
        if str(row.get("agent_direction", "") or "").upper() == macro_direction
    ][-max(1, lookback) :]
    if not relevant_rows:
        result["state"] = "no_samples"
        return result

    aligned_rows = [row for row in relevant_rows if bool(row.get("agent_aligned", False))]
    counter_rows = [row for row in relevant_rows if not bool(row.get("agent_aligned", False))]
    aligned_hit_rate = _hit_rate(aligned_rows)
    counter_hit_rate = _hit_rate(counter_rows)
    aligned_loss_streak = _loss_streak(aligned_rows)
    counter_delta = float(guard_policy.get("counter_win_bias_delta", 0.15) or 0.15)
    counter_edge = (
        len(counter_rows) >= min_samples
        and counter_hit_rate >= min(1.0, aligned_hit_rate + counter_delta)
    )
    soft_threshold = int(float(guard_policy.get("aligned_loss_streak_soft", 2) or 2))
    hard_threshold = int(float(guard_policy.get("aligned_loss_streak_hard", 3) or 3))
    soft_scale = _bounded_float(guard_policy.get("soft_scale", 0.55), 0.55, low=0.0, high=1.0)
    hard_scale = _bounded_float(guard_policy.get("hard_scale", 0.2), 0.2, low=0.0, high=1.0)
    neutralize_on_hard = _coerce_bool(guard_policy.get("neutralize_on_hard", True), True)
    scale = 1.0
    state = "normal"
    reason = ""
    if aligned_loss_streak >= hard_threshold:
        scale = hard_scale
        state = "hard_penalty"
        reason = f"aligned_loss_streak:{aligned_loss_streak}"
    elif counter_edge and len(aligned_rows) >= min_samples and aligned_hit_rate <= 0.5:
        scale = hard_scale
        state = "hard_penalty"
        reason = f"counter_edge:{round(counter_hit_rate, 3)}>{round(aligned_hit_rate, 3)}"
    elif aligned_loss_streak >= soft_threshold:
        scale = soft_scale
        state = "soft_penalty"
        reason = f"aligned_loss_streak:{aligned_loss_streak}"
    elif counter_edge:
        scale = soft_scale
        state = "soft_penalty"
        reason = f"counter_edge:{round(counter_hit_rate, 3)}>{round(aligned_hit_rate, 3)}"
    effective_direction = str(macro_state.get("effective_direction", macro_direction) or macro_direction).upper()
    disempowered = False
    if state == "hard_penalty" and neutralize_on_hard:
        effective_direction = "NEUTRAL"
        disempowered = True
    result.update(
        {
            "state": state,
            "reason": reason,
            "scale": round(scale, 4),
            "sample_count": len(relevant_rows),
            "aligned_count": len(aligned_rows),
            "counter_count": len(counter_rows),
            "aligned_hit_rate": round(aligned_hit_rate, 4),
            "counter_hit_rate": round(counter_hit_rate, 4),
            "aligned_loss_streak": aligned_loss_streak,
            "effective_direction": effective_direction,
            "disempowered": disempowered,
        }
    )
    return result


def _compute_macro_bias(macro_payload: Dict[str, Any], policy: Dict[str, Any]) -> Dict[str, Any]:
    return build_macro_direction_snapshot(
        macro_payload,
        (policy.get("macro", {}) if isinstance(policy.get("macro", {}), dict) else {}),
    )


def _build_strategy_route(agent_context: Dict[str, Any], engine_request: Dict[str, Any]) -> Dict[str, Any]:
    macro = agent_context.get("macro", {}) if isinstance(agent_context.get("macro", {}), dict) else {}
    interpreter = agent_context.get("interpreter", {}) if isinstance(agent_context.get("interpreter", {}), dict) else {}
    policy = (
        agent_context.get("decision_policy", {})
        if isinstance(agent_context.get("decision_policy", {}), dict)
        else {}
    )
    router = policy.get("router", {}) if isinstance(policy.get("router", {}), dict) else {}
    plan = engine_request.get("ai_trade_plan", {}) if isinstance(engine_request.get("ai_trade_plan", {}), dict) else {}

    regime = _normalize_regime_label(
        interpreter.get("market_structure")
        or plan.get("regime")
        or plan.get("mode")
        or "UNCERTAIN"
    )
    subtype = _normalize_regime_label(interpreter.get("regime_subtype") or plan.get("regime_subtype") or "")

    if subtype in TRENDING_SUBTYPES or regime in TRENDING_REGIMES:
        route_family = "trend"
        route_config = router.get("trend", {}) if isinstance(router.get("trend", {}), dict) else {}
        default_primary = "ema_macd"
        default_supports = ["composite", "elliott_wave"]
    elif subtype in RANGING_SUBTYPES or regime in RANGING_REGIMES:
        route_family = "range"
        route_config = router.get("range", {}) if isinstance(router.get("range", {}), dict) else {}
        default_primary = "wyckoff"
        default_supports = ["composite", "chanlun"]
    else:
        route_family = "uncertain"
        route_config = router.get("uncertain", {}) if isinstance(router.get("uncertain", {}), dict) else {}
        default_primary = "composite"
        default_supports = ["ema_macd", "wyckoff"]
    normalized_family = _normalize_router_family(
        route_config,
        primary_strategy_id=default_primary,
        support_strategy_ids=default_supports,
        rank_weights=[0.6, 0.25, 0.15],
    )
    primary = normalized_family["primary_strategy_id"]
    supports = normalized_family["support_strategy_ids"]

    effective_direction = str(macro.get("effective_direction", "NEUTRAL") or "NEUTRAL").upper()
    macro_authority = _resolve_macro_authority_context(macro)
    binding_direction = str(
        macro.get("binding_direction", macro.get("direction", effective_direction))
        or "NEUTRAL"
    ).upper()
    allowed_side = resolve_allowed_side(binding_direction, macro_authority)

    return {
        "route_family": route_family,
        "regime": regime,
        "regime_subtype": subtype,
        "primary_strategy_id": primary,
        "candidate_strategy_ids": [primary, *supports],
        "rank_weights": list(normalized_family["rank_weights"]),
        "allowed_side": allowed_side,
        "macro_authority": macro_authority,
        "macro_state": str(macro.get("state", "UNKNOWN") or "UNKNOWN"),
        "macro_direction": str(macro.get("direction", "NEUTRAL") or "NEUTRAL").upper(),
        "effective_macro_direction": effective_direction,
        "binding_macro_direction": binding_direction,
        "direction_guard": (
            macro.get("direction_guard", {})
            if isinstance(macro.get("direction_guard", {}), dict)
            else {}
        ),
        "reason": f"{route_family}:{regime}:{subtype or 'NO_SUBTYPE'}:{allowed_side}",
    }


def _route_scheduler_snapshot(
    scheduler_snapshot: Dict[str, Any],
    agent_context: Dict[str, Any],
    engine_request: Dict[str, Any],
) -> Dict[str, Any]:
    route = _build_strategy_route(agent_context, engine_request)
    instances = scheduler_snapshot.get("instances", []) if isinstance(scheduler_snapshot.get("instances", []), list) else []
    if len(instances) <= 1:
        routed = dict(scheduler_snapshot)
        routed["route"] = route
        return routed

    available_by_strategy: Dict[str, list[Dict[str, Any]]] = {}
    for instance in instances:
        if not isinstance(instance, dict):
            continue
        strategy_id = str(instance.get("strategy_id", "") or "").strip()
        if not strategy_id:
            continue
        available_by_strategy.setdefault(strategy_id, []).append(dict(instance))

    ordered_strategies = [
        strategy_id
        for strategy_id in route["candidate_strategy_ids"]
        if strategy_id in available_by_strategy
    ]
    if not ordered_strategies:
        # Agent route targets strategies outside the active set (e.g. wyckoff
        # for RANGE when only ema_macd+elliott_wave are enabled). Fall back to
        # available strategies so the agent's market read still drives selection
        # rather than preserving a stale hardcoded selected_instance_id.
        available_ids = list(available_by_strategy.keys())
        if not available_ids:
            routed = dict(scheduler_snapshot)
            routed["route"] = {**route, "fallback": "no_candidate_available"}
            return routed
        ordered_strategies = available_ids
        route["candidate_strategy_ids"] = available_ids
        route["rank_weights"] = []

    configured_candidates = (
        route.get("candidate_strategy_ids", [])
        if isinstance(route.get("candidate_strategy_ids", []), list)
        else []
    )
    configured_rank_weights = (
        route.get("rank_weights", [])
        if isinstance(route.get("rank_weights", []), list)
        else []
    )
    strategy_route_weights: Dict[str, float] = {}
    for strategy_id in ordered_strategies:
        try:
            configured_index = configured_candidates.index(strategy_id)
        except ValueError:
            configured_index = -1
        configured_weight = (
            configured_rank_weights[configured_index]
            if 0 <= configured_index < len(configured_rank_weights)
            else 0.0
        )
        strategy_route_weights[strategy_id] = round(max(0.0, _safe_float(configured_weight, 0.0)), 6)
    resolved_rank_weights = _resolve_rank_weights(configured_rank_weights, len(ordered_strategies))
    execution_strategy_weights = {
        strategy_id: resolved_rank_weights[idx]
        for idx, strategy_id in enumerate(ordered_strategies)
    }
    if not any(weight > 0.0 for weight in strategy_route_weights.values()):
        strategy_route_weights = dict(execution_strategy_weights)

    routed_instances: list[Dict[str, Any]] = []
    instance_weights: Dict[str, float] = {}
    strategy_weights: Dict[str, float] = {}
    strategy_instance_map: Dict[str, list[str]] = {}

    for strategy_id in ordered_strategies:
        rows = available_by_strategy.get(strategy_id, [])
        if not rows:
            continue
        base_weight = float(execution_strategy_weights.get(strategy_id, 0.0) or 0.0)
        row_total = sum(max(0.0, float(row.get("weight", row.get("allocation", 0.0)) or 0.0)) for row in rows)
        if row_total <= 0:
            row_total = float(len(rows))
        for row in rows:
            routed_row = dict(row)
            instance_id = str(routed_row.get("instance_id", "") or "")
            row_weight = max(0.0, float(routed_row.get("weight", routed_row.get("allocation", 0.0)) or 0.0))
            if row_weight <= 0:
                row_weight = 1.0
            normalized_weight = base_weight * (row_weight / row_total)
            routed_row["weight"] = round(normalized_weight, 6)
            routed_instances.append(routed_row)
            instance_weights[instance_id] = round(normalized_weight, 6)
            strategy_instance_map.setdefault(strategy_id, []).append(instance_id)
        strategy_weights[strategy_id] = round(base_weight, 6)

    if not routed_instances:
        routed = dict(scheduler_snapshot)
        routed["route"] = {**route, "fallback": "empty_routed_instances"}
        return routed

    selected_instance_id = max(instance_weights.items(), key=lambda item: (float(item[1]), item[0]))[0]
    selected_strategy_id = ""
    for strategy_id, instance_ids in strategy_instance_map.items():
        if selected_instance_id in instance_ids:
            selected_strategy_id = strategy_id
            break

    routed = dict(scheduler_snapshot)
    route["weights"] = strategy_route_weights
    route["routed_strategy_ids"] = list(ordered_strategies)
    route["entry_policy"] = "selected_strategy_only"
    route["entry_owner_instance_id"] = selected_instance_id
    route["entry_owner_strategy_id"] = selected_strategy_id or ordered_strategies[0]
    routed.update(
        {
            "mode": "single_active" if len(routed_instances) == 1 else "multi_active",
            "arbiter": "route-weighted",
            "reason": f"strategy_router:{route['reason']}",
            "active_instance_ids": [str(item.get("instance_id", "") or "") for item in routed_instances],
            "active_strategy_ids": ordered_strategies,
            "instance_weights": instance_weights,
            "strategy_weights": strategy_weights,
            "strategy_instance_map": strategy_instance_map,
            "instances": routed_instances,
            "selected_instance_id": selected_instance_id,
            "selected_strategy_id": selected_strategy_id or ordered_strategies[0],
            "route": route,
        }
    )
    return routed


def _resolve_binding_direction(macro_context: Dict[str, Any]) -> str:
    binding_direction = str(
        macro_context.get(
            "binding_direction",
            macro_context.get("direction", macro_context.get("effective_direction", "NEUTRAL")),
        )
        or "NEUTRAL"
    ).upper()
    if binding_direction not in {"LONG", "SHORT", "NEUTRAL"}:
        return "NEUTRAL"
    return binding_direction


def _resolve_macro_authority_context(macro_context: Dict[str, Any]) -> str:
    authority = str(macro_context.get("macro_authority", "") or "").upper()
    direction_guard = (
        macro_context.get("direction_guard", {})
        if isinstance(macro_context.get("direction_guard", {}), dict)
        else {}
    )
    guard_state = str(direction_guard.get("state", "inactive") or "inactive").lower()
    disempowered = bool(macro_context.get("direction_disempowered", False))
    if guard_state == "hard_penalty":
        disempowered = True
    if authority == "DISABLED" or disempowered:
        return "DISABLED"
    if authority == "SOFT_BIAS" or guard_state == "soft_penalty":
        return "SOFT_BIAS"
    if authority == "HARD_BIND" and guard_state not in {"soft_penalty", "hard_penalty"}:
        return "HARD_BIND"
    authority = resolve_macro_authority(
        str(
            macro_context.get(
                "direction",
                macro_context.get("binding_direction", macro_context.get("effective_direction", "NEUTRAL")),
            )
            or "NEUTRAL"
        ),
        state=str(macro_context.get("state", "UNKNOWN") or "UNKNOWN"),
        degraded=bool(macro_context.get("degraded", False)),
        bias_weight=_safe_float(macro_context.get("bias_weight", 0.0), 0.0),
        disempowered=disempowered,
    )
    if authority == "HARD_BIND" and guard_state == "soft_penalty":
        return "SOFT_BIAS"
    return authority


def _resolve_execution_binding_direction(macro_context: Dict[str, Any]) -> str:
    if _resolve_macro_authority_context(macro_context) != "HARD_BIND":
        return "NEUTRAL"
    return _resolve_binding_direction(macro_context)


def _binding_side_conflicts(binding_direction: str, signal_side: str) -> bool:
    binding_upper = str(binding_direction or "NEUTRAL").upper()
    side_upper = str(signal_side or "NONE").upper()
    return (
        (binding_upper == "LONG" and side_upper == "SELL")
        or (binding_upper == "SHORT" and side_upper == "BUY")
    )


def _structural_side_conflicts(structural_side: str, signal_side: str) -> bool:
    structural_upper = str(structural_side or "NONE").upper()
    side_upper = str(signal_side or "NONE").upper()
    return structural_upper in {"BUY", "SELL"} and side_upper in {"BUY", "SELL"} and structural_upper != side_upper


def _resolve_structural_side(
    *,
    structural_side: str,
    dominant_direction: str,
    continuation_bias: str,
) -> str:
    side_upper = str(structural_side or "").upper()
    if side_upper in {"BUY", "SELL"}:
        return side_upper
    dominant_upper = str(dominant_direction or "").upper()
    continuation_upper = str(continuation_bias or "").upper()
    if dominant_upper == "BULLISH" and continuation_upper != "BEARISH":
        return "BUY"
    if dominant_upper == "BEARISH" and continuation_upper != "BULLISH":
        return "SELL"
    if continuation_upper == "BULLISH" and dominant_upper != "BEARISH":
        return "BUY"
    if continuation_upper == "BEARISH" and dominant_upper != "BULLISH":
        return "SELL"
    return "NONE"


def _build_structural_hard_conflict_guard(
    macro_context: Dict[str, Any],
    interpreter_context: Dict[str, Any],
) -> Dict[str, Any]:
    binding_direction = _resolve_binding_direction(macro_context)
    macro_authority = _resolve_macro_authority_context(macro_context)
    dominant_direction = str(interpreter_context.get("dominant_direction", "") or "").upper()
    continuation_bias = str(interpreter_context.get("continuation_bias", "") or "").upper()
    structural_side = _resolve_structural_side(
        structural_side=str(interpreter_context.get("structural_side", "") or ""),
        dominant_direction=dominant_direction,
        continuation_bias=continuation_bias,
    )
    execution_advice = str(interpreter_context.get("execution_advice", "ALLOW_FOLLOW") or "ALLOW_FOLLOW").upper()
    counter_trend_risk = str(interpreter_context.get("counter_trend_risk", "LOW") or "LOW").upper()
    same_side_votes = 0
    opposite_side_votes = 0
    if structural_side == "BUY":
        same_side_votes += int(dominant_direction == "BULLISH")
        same_side_votes += int(continuation_bias == "BULLISH")
        opposite_side_votes += int(dominant_direction == "BEARISH")
        opposite_side_votes += int(continuation_bias == "BEARISH")
    elif structural_side == "SELL":
        same_side_votes += int(dominant_direction == "BEARISH")
        same_side_votes += int(continuation_bias == "BEARISH")
        opposite_side_votes += int(dominant_direction == "BULLISH")
        opposite_side_votes += int(continuation_bias == "BULLISH")
    conflict_with_binding = _binding_side_conflicts(binding_direction, structural_side)
    macro_hard_bind_active = macro_authority == "HARD_BIND" and binding_direction in {"LONG", "SHORT"}
    ready = (
        structural_side in {"BUY", "SELL"}
        and execution_advice in STRUCTURAL_HARD_BLOCK_ADVICES
        and not macro_hard_bind_active
    )
    return {
        "enabled": ready,
        "macro_authority": macro_authority,
        "binding_direction": binding_direction,
        "structural_side": structural_side,
        "dominant_direction": dominant_direction,
        "continuation_bias": continuation_bias,
        "execution_advice": execution_advice,
        "counter_trend_risk": counter_trend_risk,
        "same_side_votes": same_side_votes,
        "opposite_side_votes": opposite_side_votes,
        "reason": (
            f"structural_side={structural_side or 'NONE'} binding={binding_direction} "
            f"authority={macro_authority} advice={execution_advice} risk={counter_trend_risk}"
        ),
    }


def _build_binding_rejection_outcome(
    ctx: PipelineRunContext,
    *,
    binding_direction: str,
    block_reason: str,
) -> PipelineOutcome:
    side = str(ctx.signal.get("side", "NONE") or "NONE").upper()
    ctx.signal["reject_reason"] = block_reason
    ctx.signal["binding_direction"] = binding_direction
    return PipelineOutcome(
        execution_report=_build_execution_report(
            ctx.signal_id,
            "REJECTED",
            block_reason,
            action=str(ctx.signal.get("action", "HOLD") or "HOLD").upper(),
            side=side,
            binding_direction=binding_direction,
        ),
        ledger_events=["macro_binding_blocked"],
        append_runtime_log=True,
    )


def _build_open_guard_rejection_outcome(
    ctx: PipelineRunContext,
    *,
    block_reason: str,
    guard_snapshot: Dict[str, Any],
) -> PipelineOutcome:
    side = str(ctx.signal.get("side", "NONE") or "NONE").upper()
    ctx.signal["reject_reason"] = block_reason
    ctx.signal["open_guard"] = dict(guard_snapshot)
    return PipelineOutcome(
        execution_report=_build_execution_report(
            ctx.signal_id,
            "REJECTED",
            block_reason,
            action=str(ctx.signal.get("action", "HOLD") or "HOLD").upper(),
            side=side,
            open_guard=dict(guard_snapshot),
        ),
        ledger_events=["runtime_open_guard_blocked"],
        append_runtime_log=True,
    )


def _default_decision_policy() -> Dict[str, Any]:
    return {
        "policy_version": "1",
        "profile": "agent_guided_multi_strategy",
        "macro": {
            "role": "directional_gate",
            "degraded_policy": "soften_to_neutral",
            "aligned_scale": 1.1,
            "countertrend_scale": 0.35,
            "neutral_scale": 1.0,
            "direction_loss_guard": {
                "enabled": True,
                "lookback": 8,
                "min_samples": 3,
                "aligned_loss_streak_soft": 2,
                "aligned_loss_streak_hard": 3,
                "counter_win_bias_delta": 0.15,
                "soft_scale": 0.55,
                "hard_scale": 0.2,
                "neutralize_on_hard": True,
            },
        },
        "interpreter": {
            "role": "execution_bias_only",
            "buy_scale": 1.0,
            "sell_scale": 1.0,
            "limit_countertrend_scale": 0.5,
            "block_countertrend_scale": 0.0,
        },
        "risk": {
            "role": "active_governor",
            "per_instance": {
                "health_scoring": {
                    "enabled": True,
                    "min_trades_for_scoring": 3,
                    "score_ttl_sec": 30,
                },
                "loss_guard": {
                    "enabled": True,
                    "soft_loss_streak": 2,
                    "hard_loss_streak": 4,
                    "shadow_loss_streak": 5,
                    "soft_scale": 0.55,
                    "hard_scale": 0.25,
                    "rehab_min_trades": 5,
                    "rehab_min_win_rate": 0.50,
                },
                "regime_fitness": {
                    "enabled": True,
                    "min_samples_per_regime": 2,
                    "unfamiliar_regime_scale": 0.6,
                },
            },
            "cross_strategy": {
                "correlation_tracking": {
                    "enabled": True,
                    "divergence_confidence_penalty": 0.4,
                    "divergence_volume_penalty": 0.6,
                    "divergence_cooldown_sec": 120,
                    "alignment_volume_bonus": 0.05,
                },
            },
            "risk_budget": {
                "allocation_method": "health_squared",
                "min_lot_per_instance": 0.01,
                "max_lot_per_instance": 0.20,
            },
        },
        "execution": {
            "aggregation": "weighted_net_confidence",
            "min_confidence": 0.15,
            "entry_policy": "health_weighted_vote",
            "advice_matrix": _default_execution_advice_matrix(),
        },
        "router": {
            "trend": {
                "primary_strategy_id": "ema_macd",
                "support_strategy_ids": ["composite", "elliott_wave"],
                "rank_weights": [0.6, 0.25, 0.15],
            },
            "range": {
                "primary_strategy_id": "wyckoff",
                "support_strategy_ids": ["composite", "chanlun"],
                "rank_weights": [0.6, 0.25, 0.15],
            },
            "uncertain": {
                "primary_strategy_id": "composite",
                "support_strategy_ids": ["ema_macd", "wyckoff"],
                "rank_weights": [0.6, 0.25, 0.15],
            },
        },
    }


def _normalize_decision_policy(raw: Dict[str, Any]) -> Dict[str, Any]:
    base = _default_decision_policy()
    payload = raw if isinstance(raw, dict) else {}
    macro = payload.get("macro", {}) if isinstance(payload.get("macro", {}), dict) else {}
    interpreter = payload.get("interpreter", {}) if isinstance(payload.get("interpreter", {}), dict) else {}
    risk = payload.get("risk", {}) if isinstance(payload.get("risk", {}), dict) else {}
    execution = payload.get("execution", {}) if isinstance(payload.get("execution", {}), dict) else {}
    router = payload.get("router", {}) if isinstance(payload.get("router", {}), dict) else {}
    macro_guard = (
        macro.get("direction_loss_guard", {})
        if isinstance(macro.get("direction_loss_guard", {}), dict)
        else {}
    )

    return {
        "policy_version": str(payload.get("policy_version", base["policy_version"]) or base["policy_version"]),
        "profile": str(payload.get("profile", base["profile"]) or base["profile"]),
        "macro": {
            "role": str(macro.get("role", base["macro"]["role"]) or base["macro"]["role"]),
            "degraded_policy": str(
                macro.get("degraded_policy", base["macro"]["degraded_policy"])
                or base["macro"]["degraded_policy"]
            ),
            "aligned_scale": _bounded_float(
                macro.get("aligned_scale", base["macro"]["aligned_scale"]),
                base["macro"]["aligned_scale"],
            ),
            "countertrend_scale": _bounded_float(
                macro.get("countertrend_scale", base["macro"]["countertrend_scale"]),
                base["macro"]["countertrend_scale"],
            ),
            "neutral_scale": _bounded_float(
                macro.get("neutral_scale", base["macro"]["neutral_scale"]),
                base["macro"]["neutral_scale"],
            ),
            "direction_loss_guard": {
                "enabled": _coerce_bool(
                    macro_guard.get("enabled", base["macro"]["direction_loss_guard"]["enabled"]),
                    base["macro"]["direction_loss_guard"]["enabled"],
                ),
                "lookback": int(
                    max(
                        1,
                        min(
                            40,
                            int(float(macro_guard.get("lookback", base["macro"]["direction_loss_guard"]["lookback"]) or base["macro"]["direction_loss_guard"]["lookback"])),
                        ),
                    )
                ),
                "min_samples": int(
                    max(
                        1,
                        min(
                            20,
                            int(float(macro_guard.get("min_samples", base["macro"]["direction_loss_guard"]["min_samples"]) or base["macro"]["direction_loss_guard"]["min_samples"])),
                        ),
                    )
                ),
                "aligned_loss_streak_soft": int(
                    max(
                        1,
                        min(
                            10,
                            int(float(macro_guard.get("aligned_loss_streak_soft", base["macro"]["direction_loss_guard"]["aligned_loss_streak_soft"]) or base["macro"]["direction_loss_guard"]["aligned_loss_streak_soft"])),
                        ),
                    )
                ),
                "aligned_loss_streak_hard": int(
                    max(
                        1,
                        min(
                            10,
                            int(float(macro_guard.get("aligned_loss_streak_hard", base["macro"]["direction_loss_guard"]["aligned_loss_streak_hard"]) or base["macro"]["direction_loss_guard"]["aligned_loss_streak_hard"])),
                        ),
                    )
                ),
                "counter_win_bias_delta": _bounded_float(
                    macro_guard.get("counter_win_bias_delta", base["macro"]["direction_loss_guard"]["counter_win_bias_delta"]),
                    base["macro"]["direction_loss_guard"]["counter_win_bias_delta"],
                    low=0.0,
                    high=1.0,
                ),
                "soft_scale": _bounded_float(
                    macro_guard.get("soft_scale", base["macro"]["direction_loss_guard"]["soft_scale"]),
                    base["macro"]["direction_loss_guard"]["soft_scale"],
                    low=0.0,
                    high=1.0,
                ),
                "hard_scale": _bounded_float(
                    macro_guard.get("hard_scale", base["macro"]["direction_loss_guard"]["hard_scale"]),
                    base["macro"]["direction_loss_guard"]["hard_scale"],
                    low=0.0,
                    high=1.0,
                ),
                "neutralize_on_hard": _coerce_bool(
                    macro_guard.get("neutralize_on_hard", base["macro"]["direction_loss_guard"]["neutralize_on_hard"]),
                    base["macro"]["direction_loss_guard"]["neutralize_on_hard"],
                ),
            },
        },
        "interpreter": {
            "role": str(interpreter.get("role", base["interpreter"]["role"]) or base["interpreter"]["role"]),
            "buy_scale": _bounded_float(
                interpreter.get("buy_scale", base["interpreter"]["buy_scale"]),
                base["interpreter"]["buy_scale"],
            ),
            "sell_scale": _bounded_float(
                interpreter.get("sell_scale", base["interpreter"]["sell_scale"]),
                base["interpreter"]["sell_scale"],
            ),
            "limit_countertrend_scale": _bounded_float(
                interpreter.get(
                    "limit_countertrend_scale",
                    base["interpreter"]["limit_countertrend_scale"],
                ),
                base["interpreter"]["limit_countertrend_scale"],
            ),
            "block_countertrend_scale": _bounded_float(
                interpreter.get(
                    "block_countertrend_scale",
                    base["interpreter"]["block_countertrend_scale"],
                ),
                base["interpreter"]["block_countertrend_scale"],
            ),
        },
        "risk": {
            "role": str(risk.get("role", base["risk"]["role"]) or base["risk"]["role"]),
            "per_instance": {
                "health_scoring": {
                    "enabled": _coerce_bool(
                        risk.get("per_instance", {}).get("health_scoring", {}).get("enabled", True),
                        True,
                    ),
                    "min_trades_for_scoring": int(float(
                        risk.get("per_instance", {}).get("health_scoring", {}).get("min_trades_for_scoring", 3) or 3
                    )),
                    "score_ttl_sec": int(float(
                        risk.get("per_instance", {}).get("health_scoring", {}).get("score_ttl_sec", 30) or 30
                    )),
                },
                "loss_guard": {
                    "enabled": _coerce_bool(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("enabled", True),
                        True,
                    ),
                    "soft_loss_streak": int(float(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("soft_loss_streak", 2) or 2
                    )),
                    "hard_loss_streak": int(float(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("hard_loss_streak", 4) or 4
                    )),
                    "shadow_loss_streak": int(float(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("shadow_loss_streak", 5) or 5
                    )),
                    "soft_scale": _bounded_float(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("soft_scale", 0.55),
                        0.55, low=0.0, high=1.0,
                    ),
                    "hard_scale": _bounded_float(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("hard_scale", 0.25),
                        0.25, low=0.0, high=1.0,
                    ),
                    "rehab_min_trades": int(float(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("rehab_min_trades", 5) or 5
                    )),
                    "rehab_min_win_rate": _bounded_float(
                        risk.get("per_instance", {}).get("loss_guard", {}).get("rehab_min_win_rate", 0.50),
                        0.50, low=0.0, high=1.0,
                    ),
                },
                "regime_fitness": {
                    "enabled": _coerce_bool(
                        risk.get("per_instance", {}).get("regime_fitness", {}).get("enabled", True),
                        True,
                    ),
                    "min_samples_per_regime": int(float(
                        risk.get("per_instance", {}).get("regime_fitness", {}).get("min_samples_per_regime", 2) or 2
                    )),
                    "unfamiliar_regime_scale": _bounded_float(
                        risk.get("per_instance", {}).get("regime_fitness", {}).get("unfamiliar_regime_scale", 0.6),
                        0.6, low=0.0, high=1.0,
                    ),
                },
            },
            "cross_strategy": {
                "correlation_tracking": {
                    "enabled": _coerce_bool(
                        risk.get("cross_strategy", {}).get("correlation_tracking", {}).get("enabled", True),
                        True,
                    ),
                    "divergence_confidence_penalty": _bounded_float(
                        risk.get("cross_strategy", {}).get("correlation_tracking", {}).get("divergence_confidence_penalty", 0.4),
                        0.4, low=0.0, high=1.0,
                    ),
                    "divergence_volume_penalty": _bounded_float(
                        risk.get("cross_strategy", {}).get("correlation_tracking", {}).get("divergence_volume_penalty", 0.6),
                        0.6, low=0.0, high=1.0,
                    ),
                    "divergence_cooldown_sec": int(float(
                        risk.get("cross_strategy", {}).get("correlation_tracking", {}).get("divergence_cooldown_sec", 120) or 120
                    )),
                    "alignment_volume_bonus": _bounded_float(
                        risk.get("cross_strategy", {}).get("correlation_tracking", {}).get("alignment_volume_bonus", 0.05),
                        0.05, low=0.0, high=0.5,
                    ),
                },
            },
            "risk_budget": {
                "allocation_method": str(
                    risk.get("risk_budget", {}).get("allocation_method", "health_squared") or "health_squared"
                ),
                "min_lot_per_instance": _bounded_float(
                    risk.get("risk_budget", {}).get("min_lot_per_instance", 0.01),
                    0.01, low=0.0, high=0.10,
                ),
                "max_lot_per_instance": _bounded_float(
                    risk.get("risk_budget", {}).get("max_lot_per_instance", 0.20),
                    0.20, low=0.0, high=10.0,
                ),
            },
        },
        "execution": {
            "aggregation": str(
                execution.get("aggregation", base["execution"]["aggregation"])
                or base["execution"]["aggregation"]
            ),
            "min_confidence": _bounded_float(
                execution.get("min_confidence", base["execution"]["min_confidence"]),
                base["execution"]["min_confidence"],
                low=0.0,
                high=1.0,
            ),
            "entry_policy": str(
                execution.get("entry_policy", base["execution"]["entry_policy"])
                or base["execution"]["entry_policy"]
            ),
            "advice_matrix": _normalize_execution_advice_matrix(
                execution.get("advice_matrix", base["execution"]["advice_matrix"])
            ),
        },
        "router": {
            "trend": _normalize_router_family(
                router.get("trend", {}),
                primary_strategy_id=base["router"]["trend"]["primary_strategy_id"],
                support_strategy_ids=base["router"]["trend"]["support_strategy_ids"],
                rank_weights=base["router"]["trend"]["rank_weights"],
            ),
            "range": _normalize_router_family(
                router.get("range", {}),
                primary_strategy_id=base["router"]["range"]["primary_strategy_id"],
                support_strategy_ids=base["router"]["range"]["support_strategy_ids"],
                rank_weights=base["router"]["range"]["rank_weights"],
            ),
            "uncertain": _normalize_router_family(
                router.get("uncertain", {}),
                primary_strategy_id=base["router"]["uncertain"]["primary_strategy_id"],
                support_strategy_ids=base["router"]["uncertain"]["support_strategy_ids"],
                rank_weights=base["router"]["uncertain"]["rank_weights"],
            ),
        },
    }

def _load_agent_decision_context(project_root: Path, scheduler_snapshot: Dict[str, Any]) -> Dict[str, Any]:
    contract_payload = load_agent_decision_contract()
    policy_source = (
        scheduler_snapshot.get("decision_policy", {})
        if isinstance(scheduler_snapshot.get("decision_policy", {}), dict)
        and scheduler_snapshot.get("decision_policy", {})
        else contract_payload.get("decision_policy", {})
    )
    policy = _normalize_decision_policy(policy_source)
    macro_payload = (
        contract_payload.get("macro", {})
        if isinstance(contract_payload.get("macro", {}), dict)
        else {}
    )
    interpreter_payload = (
        contract_payload.get("interpreter", {})
        if isinstance(contract_payload.get("interpreter", {}), dict)
        else {}
    )
    dominant_direction = str(interpreter_payload.get("dominant_direction", "") or "").upper()
    continuation_bias = str(interpreter_payload.get("continuation_bias", "") or "").upper()
    structural_side = ""
    if dominant_direction == "BULLISH" or continuation_bias == "BULLISH":
        structural_side = "BUY"
    elif dominant_direction == "BEARISH" or continuation_bias == "BEARISH":
        structural_side = "SELL"
    macro_state = _compute_macro_bias(macro_payload, policy)
    direction_guard = _compute_direction_loss_guard(project_root, macro_state, policy)
    bias_before_guard = _safe_float(macro_state.get("bias_weight", 1.0), 1.0)
    macro_state["bias_weight_before_guard"] = round(bias_before_guard, 4)
    direction_guard_scale = _safe_float(direction_guard.get("scale", 1.0), 1.0)
    macro_state["bias_weight"] = round(
        max(0.0, min(1.0, bias_before_guard * direction_guard_scale)),
        4,
    )
    macro_state["effective_direction"] = str(
        direction_guard.get(
            "effective_direction",
            macro_state.get("effective_direction", "NEUTRAL"),
        )
        or "NEUTRAL"
    ).upper()
    macro_state["direction_guard"] = direction_guard
    macro_state["direction_disempowered"] = bool(direction_guard.get("disempowered", False))
    macro_state["macro_authority"] = _resolve_macro_authority_context(
        {
            **macro_state,
            "direction_guard": direction_guard,
        }
    )
    macro_state["allowed_side"] = resolve_allowed_side(
        str(macro_state.get("binding_direction", macro_state.get("direction", "NEUTRAL")) or "NEUTRAL"),
        str(macro_state.get("macro_authority", "DISABLED") or "DISABLED"),
    )
    # ── 写入 direction_quality 到 contract，让 Commander 感知被架空状态 ──
    if macro_state["direction_disempowered"]:
        try:
            from agent_decision_contract import update_macro_direction_quality
            update_macro_direction_quality({
                "disempowered": True,
                "disempower_reason": str(direction_guard.get("reason", "") or ""),
                "aligned_loss_streak": int(direction_guard.get("aligned_loss_streak", 0) or 0),
                "aligned_hit_rate": float(direction_guard.get("aligned_hit_rate", 0.0) or 0.0),
                "counter_hit_rate": float(direction_guard.get("counter_hit_rate", 0.0) or 0.0),
                "disempowered_direction": str(macro_state.get("direction", "NEUTRAL") or "NEUTRAL"),
            })
        except Exception:
            pass
    else:
        try:
            from agent_decision_contract import update_macro_direction_quality
            update_macro_direction_quality({
                "disempowered": False,
                "aligned_loss_streak": int(direction_guard.get("aligned_loss_streak", 0) or 0),
                "aligned_hit_rate": float(direction_guard.get("aligned_hit_rate", 0.0) or 0.0),
                "counter_hit_rate": float(direction_guard.get("counter_hit_rate", 0.0) or 0.0),
            })
        except Exception:
            pass
    structural_guard = _build_structural_hard_conflict_guard(
        macro_state,
        {
            "structural_side": structural_side,
            "dominant_direction": dominant_direction,
            "continuation_bias": continuation_bias,
            "execution_advice": str(interpreter_payload.get("execution_advice", "ALLOW_FOLLOW") or "ALLOW_FOLLOW").upper(),
            "counter_trend_risk": str(interpreter_payload.get("counter_trend_risk", "LOW") or "LOW").upper(),
        },
    )
    return {
        "decision_policy": policy,
        "contract": {
            "updated_at": str(contract_payload.get("updated_at", "") or ""),
            "lineage": (
                contract_payload.get("lineage", {})
                if isinstance(contract_payload.get("lineage", {}), dict)
                else {}
            ),
        },
        "macro": macro_state,
        "interpreter": {
            "execution_advice": str(interpreter_payload.get("execution_advice", "ALLOW_FOLLOW") or "ALLOW_FOLLOW").upper(),
            "buy_multiplier": float(interpreter_payload.get("buy_multiplier", 1.0) or 1.0),
            "sell_multiplier": float(interpreter_payload.get("sell_multiplier", 1.0) or 1.0),
            "structural_side": structural_side,
            "dominant_direction": dominant_direction,
            "continuation_bias": continuation_bias,
            "market_structure": _normalize_regime_label(interpreter_payload.get("market_structure", "")),
            "regime_subtype": _normalize_regime_label(interpreter_payload.get("regime_subtype", "")),
            "counter_trend_risk": str(interpreter_payload.get("counter_trend_risk", "LOW") or "LOW").upper(),
            "entry_timing_hint": str(interpreter_payload.get("entry_timing_hint", "") or "").upper(),
            "llm_confidence": float(interpreter_payload.get("llm_confidence", 0.55) or 0.55),
            "hard_conflict_guard": structural_guard,
        },
    }


def _apply_agent_decision_policy(
    strategy_signals: list[Dict[str, Any]],
    agent_context: Dict[str, Any],
) -> list[Dict[str, Any]]:
    policy = _normalize_decision_policy(agent_context.get("decision_policy", {}))
    macro_context = agent_context.get("macro", {}) if isinstance(agent_context.get("macro", {}), dict) else {}
    interpreter_context = (
        agent_context.get("interpreter", {})
        if isinstance(agent_context.get("interpreter", {}), dict)
        else {}
    )
    effective_macro_direction = str(macro_context.get("effective_direction", "NEUTRAL") or "NEUTRAL").upper()
    binding_macro_direction = _resolve_binding_direction(macro_context)
    macro_authority = _resolve_macro_authority_context(macro_context)
    macro_bias_weight = _safe_float(macro_context.get("bias_weight", 1.0), 1.0)
    macro_role = str(policy.get("macro", {}).get("role", "directional_gate") or "directional_gate")
    interpreter_role = str(policy.get("interpreter", {}).get("role", "execution_bias_only") or "execution_bias_only")
    execution_advice = str(interpreter_context.get("execution_advice", "ALLOW_FOLLOW") or "ALLOW_FOLLOW").upper()
    structural_side = str(interpreter_context.get("structural_side", "") or "").upper()
    min_confidence = float(policy.get("execution", {}).get("min_confidence", 0.15) or 0.15)

    for signal in strategy_signals:
        action = str(signal.get("action", "HOLD") or "HOLD").upper()
        side = str(signal.get("side", "NONE") or "NONE").upper()
        trace = signal.get("trace", {}) if isinstance(signal.get("trace", {}), dict) else {}
        evidence = trace.get("evidence", []) if isinstance(trace.get("evidence", []), list) else []
        raw_weight = float(signal.get("instance_weight", 0.0) or 0.0)
        adjusted_weight = raw_weight
        raw_confidence = float(trace.get("confidence", 0.0) or 0.0)
        confidence = raw_confidence
        effects: list[str] = []

        if action == "OPEN" and side in {"BUY", "SELL"}:
            if macro_role in {"directional_gate", "directional_hard_block"}:
                aligned = (
                    (effective_macro_direction == "LONG" and side == "BUY")
                    or (effective_macro_direction == "SHORT" and side == "SELL")
                )
                countertrend = (
                    (effective_macro_direction == "LONG" and side == "SELL")
                    or (effective_macro_direction == "SHORT" and side == "BUY")
                )
                if aligned:
                    aligned_scale = 1.0 + ((float(policy["macro"]["aligned_scale"]) - 1.0) * macro_bias_weight)
                    adjusted_weight *= aligned_scale
                    effects.append("macro_aligned_boost")
                elif countertrend:
                    if macro_role == "directional_hard_block":
                        adjusted_weight = 0.0
                        confidence = 0.0
                        signal["action"] = "HOLD"
                        signal["side"] = "NONE"
                        signal["reject_reason"] = str(signal.get("reject_reason", "") or "MACRO_DIRECTION_BLOCK")
                        effects.append("macro_hard_block")
                    else:
                        countertrend_scale = 1.0 - ((1.0 - float(policy["macro"]["countertrend_scale"])) * macro_bias_weight)
                        adjusted_weight *= countertrend_scale
                        effects.append("macro_countertrend_scaled")
                else:
                    adjusted_weight *= float(policy["macro"]["neutral_scale"])
            if interpreter_role == "execution_bias_only":
                base_scale = float(policy["interpreter"]["buy_scale"] if side == "BUY" else policy["interpreter"]["sell_scale"])
                runtime_scale = float(
                    interpreter_context.get("buy_multiplier", 1.0)
                    if side == "BUY"
                    else interpreter_context.get("sell_multiplier", 1.0)
                )
                # 方向性仓位倍率影响最终下单手数，不影信号存活权重。
                # 否则 Interpreter 会在权重聚合阶段就杀死逆结构信号，导致
                # 策略明明看对了方向（e.g. 技术面全面看空）却被 structural_side 碾成 HOLD。
                signal["interpreter_position_multiplier"] = round(max(0.05, base_scale * runtime_scale), 4)
                if structural_side and side != structural_side:
                    if execution_advice == "BLOCK_COUNTER_TREND":
                        adjusted_weight *= float(policy["interpreter"]["block_countertrend_scale"])
                        effects.append("interpreter_block_countertrend")
                    elif execution_advice in {"LIMIT_ONLY", "WAIT_CONFIRMATION"}:
                        adjusted_weight *= float(policy["interpreter"]["limit_countertrend_scale"])
                        effects.append("interpreter_limit_countertrend")

        adjusted_weight = round(max(0.0, adjusted_weight), 6)
        trace["confidence"] = round(max(0.0, min(0.99, confidence)), 4)
        trace["policy"] = {
            "profile": str(policy.get("profile", "") or ""),
            "macro_direction": str(macro_context.get("direction", "NEUTRAL") or "NEUTRAL"),
            "effective_macro_direction": effective_macro_direction,
            "binding_macro_direction": binding_macro_direction,
            "macro_authority": macro_authority,
            "macro_bias_weight": round(macro_bias_weight, 4),
            "macro_state": str(macro_context.get("state", "UNKNOWN") or "UNKNOWN"),
            "execution_advice": execution_advice,
            "structural_side": structural_side,
            "raw_instance_weight": round(raw_weight, 6),
            "adjusted_instance_weight": adjusted_weight,
            "effects": effects,
        }
        for item in effects:
            if item not in evidence:
                evidence.append(item)
        if action == "OPEN" and adjusted_weight <= 0.0 and not str(signal.get("reject_reason", "") or "").strip():
            signal["action"] = "HOLD"
            signal["side"] = "NONE"
            signal["reject_reason"] = "AGENT_POLICY_BLOCKED"
        trace["evidence"] = evidence
        signal["trace"] = trace
        signal["raw_instance_weight"] = round(raw_weight, 6)
        signal["instance_weight"] = adjusted_weight
        signal["_policy_effects"] = effects
        signal["_effective_min_confidence"] = min_confidence
    return strategy_signals


def _signal_signed_score(signal: Dict[str, Any], instance_weight: float) -> float:
    action = str(signal.get("action", "HOLD") or "HOLD").upper()
    side = str(signal.get("side", "NONE") or "NONE").upper()
    confidence = float((signal.get("trace", {}) or {}).get("confidence", 0.0) or 0.0)
    if action == "HOLD" or side not in {"BUY", "SELL"}:
        return 0.0
    return round(confidence * (1.0 if side == "BUY" else -1.0) * max(0.0, instance_weight), 6)


def _is_actionable_signal(signal: Dict[str, Any]) -> bool:
    action = str(signal.get("action", "HOLD") or "HOLD").upper()
    side = str(signal.get("side", "NONE") or "NONE").upper()
    return action != "HOLD" and side in {"BUY", "SELL"}


def _strategy_opportunity_to_dict(opportunity: StrategyOpportunity) -> Dict[str, Any]:
    return {
        "strategy_id": opportunity.strategy_id,
        "instance_id": opportunity.instance_id,
        "directional_view": opportunity.directional_view,
        "setup_quality": opportunity.setup_quality,
        "entry_style_preference": opportunity.entry_style_preference,
        "long_opportunity_score": round(opportunity.long_opportunity_score, 4),
        "short_opportunity_score": round(opportunity.short_opportunity_score, 4),
        "urgency": opportunity.urgency,
        "confidence": round(opportunity.confidence, 4),
        "actionable": opportunity.actionable,
        "source_action": opportunity.source_action,
        "source_side": opportunity.source_side,
        "notes": list(opportunity.notes),
    }


def _execution_constraint_to_dict(constraint: ExecutionConstraint) -> Dict[str, Any]:
    return {
        "advice": constraint.advice,
        "allowed_side": constraint.allowed_side,
        "allowed_entry_styles": list(constraint.allowed_entry_styles),
        "confirmation_required": constraint.confirmation_required,
        "probe_size_scale": round(constraint.probe_size_scale, 4),
        "structural_side": constraint.structural_side,
        "macro_direction": constraint.macro_direction,
        "block_new_open": constraint.block_new_open,
        "reason": constraint.reason,
    }


def _materialized_opportunity_to_dict(materialized: MaterializedOpportunity) -> Dict[str, Any]:
    return {
        "materialized_action": materialized.materialized_action,
        "final_action": materialized.final_action,
        "final_side": materialized.final_side,
        "entry_style": materialized.entry_style,
        "size_scale": round(materialized.size_scale, 4),
        "eligible_for_aggregation": materialized.eligible_for_aggregation,
        "reason": materialized.reason,
    }


def _opportunity_primary_score(opportunity: StrategyOpportunity) -> float:
    if opportunity.directional_view == "BUY":
        return max(0.0, float(opportunity.long_opportunity_score))
    if opportunity.directional_view == "SELL":
        return max(0.0, float(opportunity.short_opportunity_score))
    return 0.0


def _opportunity_signed_score(opportunity: StrategyOpportunity, instance_weight: float) -> float:
    score = _opportunity_primary_score(opportunity)
    if opportunity.directional_view == "BUY":
        return round(score * max(0.0, instance_weight), 6)
    if opportunity.directional_view == "SELL":
        return round(-score * max(0.0, instance_weight), 6)
    return 0.0


def _resolve_entry_style_preference(
    signal: Dict[str, Any],
    interpreter_context: Dict[str, Any],
) -> str:
    explicit_style = _normalize_entry_style(signal.get("entry_style", ""))
    if explicit_style != "MARKET" or str(signal.get("entry_style", "")).strip():
        return explicit_style
    execution_advice = str(interpreter_context.get("execution_advice", "ALLOW_FOLLOW") or "ALLOW_FOLLOW").upper()
    entry_timing_hint = str(interpreter_context.get("entry_timing_hint", "") or "").upper()
    if execution_advice == "WAIT_CONFIRMATION":
        return "CONFIRMATION_ONLY"
    if execution_advice == "LIMIT_ONLY" or entry_timing_hint == "WAIT_PULLBACK":
        return "PULLBACK_LIMIT"
    return "MARKET"


def _opportunity_setup_quality(action: str, side: str, confidence: float) -> str:
    if action != "OPEN" or side not in {"BUY", "SELL"}:
        return "INVALID"
    if confidence >= 0.8:
        return "STRONG"
    if confidence >= 0.5:
        return "VALID"
    if confidence >= 0.25:
        return "WEAK"
    return "INVALID"


def _opportunity_urgency(setup_quality: str, entry_style_preference: str, confidence: float) -> str:
    if entry_style_preference == "MARKET" and confidence >= 0.75:
        return "HIGH"
    if setup_quality in {"STRONG", "VALID"} and entry_style_preference in {"MARKET", "LIMIT"}:
        return "MEDIUM"
    return "LOW"


def _build_strategy_opportunity(
    signal: Dict[str, Any],
    interpreter_context: Dict[str, Any],
) -> StrategyOpportunity:
    action = str(signal.get("action", "HOLD") or "HOLD").upper()
    side = str(signal.get("side", "NONE") or "NONE").upper()
    confidence = float((signal.get("trace", {}) or {}).get("confidence", signal.get("confidence", 0.0)) or 0.0)
    setup_quality = _opportunity_setup_quality(action, side, confidence)
    entry_style_preference = _resolve_entry_style_preference(signal, interpreter_context)
    long_score = confidence if action == "OPEN" and side == "BUY" else 0.0
    short_score = confidence if action == "OPEN" and side == "SELL" else 0.0
    notes = []
    reject_reason = str(signal.get("reject_reason", "") or "")
    if reject_reason:
        notes.append(reject_reason)
    return StrategyOpportunity(
        strategy_id=str(signal.get("_sig_strategy_id", signal.get("strategy_id", "")) or ""),
        instance_id=str(signal.get("_instance_id", signal.get("instance_id", "")) or ""),
        directional_view=side if side in {"BUY", "SELL"} else "NONE",
        setup_quality=setup_quality,
        entry_style_preference=entry_style_preference,
        long_opportunity_score=round(long_score, 4),
        short_opportunity_score=round(short_score, 4),
        urgency=_opportunity_urgency(setup_quality, entry_style_preference, confidence),
        confidence=round(confidence, 4),
        actionable=action == "OPEN" and side in {"BUY", "SELL"},
        source_action=action,
        source_side=side,
        notes=notes,
    )


def _resolve_execution_constraint(
    agent_context: Dict[str, Any],
    route: Dict[str, Any],
    decision_policy: Dict[str, Any],
) -> ExecutionConstraint:
    macro_context = agent_context.get("macro", {}) if isinstance(agent_context.get("macro", {}), dict) else {}
    interpreter_context = (
        agent_context.get("interpreter", {})
        if isinstance(agent_context.get("interpreter", {}), dict)
        else {}
    )
    advice = str(interpreter_context.get("execution_advice", "ALLOW_FOLLOW") or "ALLOW_FOLLOW").upper()
    advice_matrix = (
        decision_policy.get("execution", {}).get("advice_matrix", {})
        if isinstance(decision_policy.get("execution", {}), dict)
        else {}
    )
    if not isinstance(advice_matrix, dict):
        advice_matrix = _default_execution_advice_matrix()
    advice_row = advice_matrix.get(advice, advice_matrix.get("ALLOW_FOLLOW", _default_execution_advice_matrix()["ALLOW_FOLLOW"]))
    allowed_side = str(
        route.get("allowed_side", macro_context.get("allowed_side", "BOTH"))
        or macro_context.get("allowed_side", "BOTH")
        or "BOTH"
    ).upper()
    structural_side = str(interpreter_context.get("structural_side", "NONE") or "NONE").upper()
    macro_direction = str(macro_context.get("effective_direction", macro_context.get("direction", "NEUTRAL")) or "NEUTRAL").upper()
    reason = f"advice={advice} allowed_side={allowed_side} structural_side={structural_side} macro={macro_direction}"
    return ExecutionConstraint(
        advice=advice,
        allowed_side=allowed_side,
        allowed_entry_styles=_normalize_entry_style_list(
            advice_row.get("allowed_entry_styles", ["MARKET", "LIMIT"]),
            ["MARKET", "LIMIT"],
        ),
        confirmation_required=_coerce_bool(advice_row.get("confirmation_required", False), False),
        probe_size_scale=_bounded_float(advice_row.get("probe_size_scale", 0.5), 0.5, low=0.1, high=1.0),
        structural_side=structural_side,
        macro_direction=macro_direction,
        block_new_open=False,
        reason=reason,
    )


def _materialize_opportunity(
    opportunity: StrategyOpportunity,
    constraint: ExecutionConstraint,
    *,
    min_confidence: float,
) -> MaterializedOpportunity:
    side = opportunity.directional_view
    score = _opportunity_primary_score(opportunity)
    allowed_styles = set(constraint.allowed_entry_styles)
    preferred_style = opportunity.entry_style_preference
    entry_style = "MARKET" if "MARKET" in allowed_styles else "LIMIT"
    if preferred_style in {"LIMIT", "PULLBACK_LIMIT"} and allowed_styles.intersection({"LIMIT", "PULLBACK_LIMIT"}):
        entry_style = "LIMIT"
    elif preferred_style == "CONFIRMATION_ONLY":
        entry_style = "LIMIT" if "LIMIT" in allowed_styles or "PULLBACK_LIMIT" in allowed_styles else "MARKET"
    if not opportunity.actionable or side not in {"BUY", "SELL"} or opportunity.setup_quality == "INVALID":
        return MaterializedOpportunity(
            materialized_action="NO_SETUP",
            reason="NO_ACTIONABLE_SETUP",
        )
    if constraint.allowed_side in {"BUY", "SELL"} and side != constraint.allowed_side:
        return MaterializedOpportunity(
            materialized_action="BLOCKED",
            reason=f"SIDE_NOT_ALLOWED:{constraint.allowed_side}",
        )
    if constraint.block_new_open:
        return MaterializedOpportunity(
            materialized_action="BLOCKED",
            reason="NEW_OPEN_BLOCKED",
        )
    if constraint.confirmation_required:
        return MaterializedOpportunity(
            materialized_action="ARM_WAIT_CONFIRMATION",
            entry_style=entry_style,
            eligible_for_aggregation=True,
            reason="WAIT_CONFIRMATION_REQUIRED",
        )
    if opportunity.setup_quality == "WEAK" and score >= max(0.0, min_confidence * 0.75):
        return MaterializedOpportunity(
            materialized_action="OPEN_PROBE",
            final_action="OPEN",
            final_side=side,
            entry_style=entry_style,
            size_scale=constraint.probe_size_scale,
            eligible_for_aggregation=True,
            reason="WEAK_SETUP_PROBE",
        )
    if score < min_confidence:
        return MaterializedOpportunity(
            materialized_action="NO_SETUP",
            reason="OPPORTUNITY_SCORE_BELOW_THRESHOLD",
        )
    return MaterializedOpportunity(
        materialized_action="OPEN_LIMIT" if entry_style == "LIMIT" else "OPEN_MARKET",
        final_action="OPEN",
        final_side=side,
        entry_style=entry_style,
        size_scale=1.0,
        eligible_for_aggregation=True,
        reason="READY_TO_EXECUTE",
    )


def _apply_price_target_overrides(
    plan: Dict[str, Any],
    param_overrides: Dict[str, Any],
) -> list[str]:
    applied: list[str] = []
    if not isinstance(param_overrides, dict) or not param_overrides:
        return applied
    if "atr_sl_mult" not in param_overrides and "tp_rr" not in param_overrides:
        return applied
    side = str(plan.get("side", "NONE") or "NONE").upper()
    if side not in {"BUY", "SELL"}:
        return applied
    try:
        entry_price = float(plan.get("entry_price", 0.0) or 0.0)
        atr_v = float(((plan.get("indicators", {}) or {}).get("atr14", 0.0) or 0.0))
    except Exception:
        return applied
    if entry_price <= 0 or atr_v <= 0:
        return applied
    try:
        sl_mult = float(param_overrides.get("atr_sl_mult", 0.0) or 0.0)
    except Exception:
        sl_mult = 0.0
    if sl_mult > 0:
        sl_distance = max(atr_v * sl_mult, 0.01)
        if side == "BUY":
            plan["sl_price"] = round(entry_price - sl_distance, 6)
        else:
            plan["sl_price"] = round(entry_price + sl_distance, 6)
        applied.append("atr_sl_mult")
    try:
        tp_rr = float(param_overrides.get("tp_rr", 0.0) or 0.0)
    except Exception:
        tp_rr = 0.0
    try:
        sl_price = float(plan.get("sl_price", 0.0) or 0.0)
    except Exception:
        sl_price = 0.0
    risk_distance = abs(entry_price - sl_price)
    if tp_rr > 0 and risk_distance > 0:
        tp_distance = max(risk_distance * tp_rr, 0.01)
        if side == "BUY":
            plan["tp_price"] = round(entry_price + tp_distance, 6)
        else:
            plan["tp_price"] = round(entry_price - tp_distance, 6)
        applied.append("tp_rr")
    return applied


def _apply_panel_sltp_to_ai_plan(
    mt5_snapshot: Dict[str, Any],
    engine_request: Dict[str, Any],
) -> None:
    plan = engine_request.get("ai_trade_plan", {}) if isinstance(engine_request.get("ai_trade_plan", {}), dict) else {}
    if not plan:
        return
    action = str(plan.get("action", "HOLD") or "HOLD").upper()
    side = str(plan.get("side", "NONE") or "NONE").upper()
    if action != "OPEN" or side not in {"BUY", "SELL"}:
        return
    try:
        entry_price = float(plan.get("entry_price", 0.0) or 0.0)
    except Exception:
        return
    if entry_price <= 0:
        market = mt5_snapshot.get("market", {}) if isinstance(mt5_snapshot.get("market", {}), dict) else {}
        try:
            entry_price = float(
                market.get("ask" if side == "BUY" else "bid", 0.0)
                or 0.0
            )
        except Exception:
            entry_price = 0.0
    if entry_price <= 0:
        return
    sl_pips = float(config_manager.get_effective_param("sl_pips", 0) or 0)
    tp_pips = float(config_manager.get_effective_param("tp_pips", 0) or 0)
    if sl_pips <= 0 or tp_pips <= 0:
        return
    price_step = 0.1
    sl_distance = round(sl_pips * price_step, 6)
    tp_distance = round(tp_pips * price_step, 6)
    if side == "BUY":
        plan["sl_price"] = round(entry_price - sl_distance, 6)
        plan["tp_price"] = round(entry_price + tp_distance, 6)
    else:
        plan["sl_price"] = round(entry_price + sl_distance, 6)
        plan["tp_price"] = round(entry_price - tp_distance, 6)
    risk_flags = plan.get("risk_flags", []) if isinstance(plan.get("risk_flags", []), list) else []
    if "PANEL_SLTP_ENFORCED" not in risk_flags:
        risk_flags.append("PANEL_SLTP_ENFORCED")
    plan["risk_flags"] = risk_flags
    engine_request["ai_trade_plan"] = plan


def _pending_entry_registry_path(project_root: Path) -> Path:
    return project_root / "data" / "pending_entry_intents.json"


def _load_pending_entry_registry(project_root: Path) -> Dict[str, Any]:
    path = _pending_entry_registry_path(project_root)
    if not path.exists():
        return {"entries": []}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {"entries": []}
    if not isinstance(payload, dict):
        return {"entries": []}
    entries = payload.get("entries", [])
    payload["entries"] = entries if isinstance(entries, list) else []
    return payload


def _save_pending_entry_registry(project_root: Path, payload: Dict[str, Any]) -> None:
    path = _pending_entry_registry_path(project_root)
    path.parent.mkdir(parents=True, exist_ok=True)
    serialized = payload if isinstance(payload, dict) else {"entries": []}
    serialized["entries"] = serialized.get("entries", []) if isinstance(serialized.get("entries", []), list) else []
    path.write_text(json.dumps(serialized, ensure_ascii=False, indent=2), encoding="utf-8")


def _is_executable_materialized_action(materialized_action: str) -> bool:
    return str(materialized_action or "").upper() in {"OPEN_MARKET", "OPEN_LIMIT", "OPEN_PROBE"}


def _market_bid_ask(mt5_snapshot: Dict[str, Any]) -> tuple[float, float]:
    market = mt5_snapshot.get("market", {}) if isinstance(mt5_snapshot.get("market", {}), dict) else {}
    try:
        bid = float(market.get("bid", 0.0) or 0.0)
    except Exception:
        bid = 0.0
    try:
        ask = float(market.get("ask", 0.0) or 0.0)
    except Exception:
        ask = 0.0
    return bid, ask


def _build_pending_entry_record(ctx: PipelineRunContext) -> Dict[str, Any]:
    signal = deepcopy(ctx.signal)
    materialized_action = str(signal.get("materialized_action", "") or "").upper()
    created_ts = time.time()
    ttl = int(PENDING_ENTRY_TTL_SECONDS.get(materialized_action, 15 * 60))
    return {
        "pending_entry_id": str(signal.get("signal_id", "") or f"pending_{int(created_ts)}"),
        "materialized_action": materialized_action,
        "entry_style": str(signal.get("entry_style", "MARKET") or "MARKET").upper(),
        "side": str(signal.get("side", "NONE") or "NONE").upper(),
        "entry_price": float(signal.get("entry_price", signal.get("price", 0.0)) or 0.0),
        "confidence": float((signal.get("trace", {}) or {}).get("confidence", signal.get("confidence", 0.0)) or 0.0),
        "strategy_id": str(signal.get("selected_strategy_id", signal.get("strategy_id", "")) or ""),
        "instance_id": str(signal.get("selected_instance_id", signal.get("instance_id", "")) or ""),
        "created_ts": created_ts,
        "expire_ts": created_ts + ttl,
        "signal": signal,
    }


def _register_pending_entry(ctx: PipelineRunContext) -> Dict[str, Any]:
    record = _build_pending_entry_record(ctx)
    registry = _load_pending_entry_registry(ctx.project_root)
    entries = registry.get("entries", []) if isinstance(registry.get("entries", []), list) else []
    filtered: list[Dict[str, Any]] = []
    for item in entries:
        if not isinstance(item, dict):
            continue
        same_materialized = str(item.get("materialized_action", "") or "").upper() == record["materialized_action"]
        same_strategy = str(item.get("strategy_id", "") or "") == record["strategy_id"]
        same_side = str(item.get("side", "") or "").upper() == record["side"]
        if same_materialized and same_strategy and same_side:
            continue
        filtered.append(item)
    filtered.append(record)
    registry["entries"] = filtered
    registry["updated_at"] = datetime.now(timezone.utc).isoformat()
    _save_pending_entry_registry(ctx.project_root, registry)
    reason = "SOFT_LIMIT_REGISTERED" if record["materialized_action"] == "OPEN_LIMIT" else "WAIT_CONFIRMATION_ARMED"
    return _build_execution_report(
        ctx.signal_id,
        "ACKED",
        reason,
        pending_entry_id=record["pending_entry_id"],
        pending_entry_kind=record["materialized_action"],
        entry_style=record["entry_style"],
        materialized_action=record["materialized_action"],
    )


def _pending_entry_triggered(entry: Dict[str, Any], mt5_snapshot: Dict[str, Any]) -> bool:
    side = str(entry.get("side", "NONE") or "NONE").upper()
    entry_price = float(entry.get("entry_price", 0.0) or 0.0)
    if side not in {"BUY", "SELL"} or entry_price <= 0:
        return False
    bid, ask = _market_bid_ask(mt5_snapshot)
    if side == "BUY":
        return ask > 0 and ask <= entry_price
    return bid > 0 and bid >= entry_price


def _pending_confirmation_resolved(entry: Dict[str, Any], current_signal: Dict[str, Any]) -> bool:
    current_action = str(current_signal.get("action", "HOLD") or "HOLD").upper()
    current_side = str(current_signal.get("side", "NONE") or "NONE").upper()
    current_materialized = str(current_signal.get("materialized_action", "") or "").upper()
    pending_side = str(entry.get("side", "NONE") or "NONE").upper()
    if current_action != "OPEN" or current_side != pending_side:
        return False
    return _is_executable_materialized_action(current_materialized)


def _apply_pending_entry_promotions(ctx: PipelineRunContext) -> None:
    registry = _load_pending_entry_registry(ctx.project_root)
    entries = registry.get("entries", []) if isinstance(registry.get("entries", []), list) else []
    if not entries:
        return
    now_ts = time.time()
    kept_entries: list[Dict[str, Any]] = []
    promoted_signal: Optional[Dict[str, Any]] = None
    promoted_reason = ""
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        if float(entry.get("expire_ts", 0.0) or 0.0) <= now_ts:
            continue
        materialized_action = str(entry.get("materialized_action", "") or "").upper()
        if materialized_action == "ARM_WAIT_CONFIRMATION":
            if _pending_confirmation_resolved(entry, ctx.signal):
                continue
            kept_entries.append(entry)
            continue
        if materialized_action == "OPEN_LIMIT":
            if promoted_signal is None and _pending_entry_triggered(entry, ctx.mt5_snapshot):
                base_signal = deepcopy(entry.get("signal", {})) if isinstance(entry.get("signal", {}), dict) else {}
                origin_signal_id = str(base_signal.get("signal_id", entry.get("pending_entry_id", "pending")) or "pending")
                base_signal["signal_id"] = f"{origin_signal_id}__soft_limit_trigger"
                base_signal["origin_signal_id"] = origin_signal_id
                base_signal["materialized_action"] = "OPEN_MARKET"
                base_signal["materialized_reason"] = "SOFT_LIMIT_TRIGGERED"
                base_signal["pending_entry_id"] = str(entry.get("pending_entry_id", "") or "")
                promoted_signal = base_signal
                promoted_reason = "SOFT_LIMIT_TRIGGERED"
                continue
            kept_entries.append(entry)
            continue
        kept_entries.append(entry)
    registry["entries"] = kept_entries
    registry["updated_at"] = datetime.now(timezone.utc).isoformat()
    _save_pending_entry_registry(ctx.project_root, registry)
    if promoted_signal is not None:
        ctx.signal = promoted_signal
        ctx.signal_id = str(promoted_signal.get("signal_id", ctx.signal_id) or ctx.signal_id)
        pending_resolution = {
            "reason": promoted_reason,
            "pending_entry_id": str(promoted_signal.get("pending_entry_id", "") or ""),
        }
        ctx.aggregated_intent["pending_entry_resolution"] = pending_resolution
        ctx.signal_attribution = _attach_signal_attribution(ctx.signal, ctx.aggregated_intent)


def _build_instance_signals(
    engine_request: Dict[str, Any],
    project_root: Path,
    scheduler_snapshot: Dict[str, Any],
) -> list[Dict[str, Any]]:
    instances = scheduler_snapshot.get("instances", []) if isinstance(scheduler_snapshot.get("instances", []), list) else []
    multiple_instances = len(instances) > 1
    current_open_positions = int(float(engine_request.get("account_metrics", {}).get("current_open_positions", 0) or 0))
    selected_entry_instance_id = str(scheduler_snapshot.get("selected_instance_id", "") or "").strip()
    route = scheduler_snapshot.get("route", {}) if isinstance(scheduler_snapshot.get("route", {}), dict) else {}
    allowed_side = str(route.get("allowed_side", "BOTH") or "BOTH").upper()
    entry_selection_active = multiple_instances and current_open_positions <= 0 and bool(selected_entry_instance_id)
    target_instances = instances
    if entry_selection_active:
        target_instances = [
            instance
            for instance in instances
            if str(instance.get("instance_id", "") or "").strip() == selected_entry_instance_id
        ]
        if not target_instances:
            target_instances = instances
    signals: list[Dict[str, Any]] = []

    def _render_instance_signal(instance: Dict[str, Any]) -> Dict[str, Any]:
        # Shallow copy + targeted deep copy of only the mutable ai_trade_plan
        # avoids expensive full deepcopy of market_snapshot/bars on every instance
        instance_request = dict(engine_request)
        raw_plan = instance_request.get("ai_trade_plan", {})
        plan = deepcopy(raw_plan) if isinstance(raw_plan, dict) else {}
        strategy_id = str(instance.get("strategy_id", "")).strip()
        instance_id = str(instance.get("instance_id", "")).strip()
        param_overrides = (
            instance.get("param_overrides", {})
            if isinstance(instance.get("param_overrides", {}), dict)
            else {}
        )
        if strategy_id:
            plan["strategy_id"] = strategy_id
        plan["_instance_id"] = instance_id
        applied_override_keys: list[str] = []
        if "risk_per_trade" in param_overrides:
            try:
                # Contract stores risk as decimal (0.01 = 1%), while executor expects pct.
                plan["risk_per_trade_pct"] = round(float(param_overrides.get("risk_per_trade", 0.0) or 0.0) * 100.0, 4)
                applied_override_keys.append("risk_per_trade")
            except Exception:
                pass
        for override_key in ("volume_lots", "entry_price", "sl_price", "tp_price"):
            if override_key not in param_overrides:
                continue
            try:
                plan[override_key] = float(param_overrides.get(override_key, 0.0) or 0.0)
                applied_override_keys.append(override_key)
            except Exception:
                continue
        if "max_positions" in param_overrides:
            try:
                plan["max_positions"] = int(float(param_overrides.get("max_positions", 0.0) or 0.0))
                applied_override_keys.append("max_positions")
            except Exception:
                pass
        applied_override_keys.extend(_apply_price_target_overrides(plan, param_overrides))
        instance_request["ai_trade_plan"] = plan
        instance_request["runtime_scheduler"] = scheduler_snapshot
        instance_request["strategy_context"] = {
            "instance_id": instance_id,
            "strategy_id": strategy_id,
            "role": str(instance.get("role", "live_support") or "live_support"),
            "param_overrides": param_overrides,
            "config_revision": str(instance.get("config_revision", scheduler_snapshot.get("config_revision", "manual"))),
        }
        signal = run_engine(instance_request, project_root)
        base_signal_id = str(signal.get("signal_id", f"sig_{strategy_id or 'strategy'}_{int(time.time())}") or f"sig_{strategy_id or 'strategy'}_{int(time.time())}")
        if multiple_instances:
            signal["signal_id"] = f"{base_signal_id}__{instance_id}"
        signal["_base_signal_id"] = base_signal_id
        signal["_instance_id"] = instance_id
        signal["_sig_strategy_id"] = strategy_id
        signal["strategy_id"] = strategy_id
        signal["config_revision"] = str(instance.get("config_revision", scheduler_snapshot.get("config_revision", "manual")))
        signal["instance_weight"] = float(scheduler_snapshot.get("instance_weights", {}).get(instance_id, instance.get("weight", 0.0)) or 0.0)
        signal["instance_role"] = str(instance.get("role", "live_support") or "live_support")
        signal["param_overrides"] = param_overrides
        signal["entry_selected"] = not entry_selection_active or instance_id == selected_entry_instance_id
        if applied_override_keys:
            trace = signal.get("trace", {}) if isinstance(signal.get("trace", {}), dict) else {}
            evidence = trace.get("evidence", []) if isinstance(trace.get("evidence", []), list) else []
            for override_key in applied_override_keys:
                marker = f"instance_override:{override_key}"
                if marker not in evidence:
                    evidence.append(marker)
            trace["evidence"] = evidence
            signal["trace"] = trace
            signal["applied_execution_overrides"] = applied_override_keys
        if bool(signal["param_overrides"].get("reduce_only", False)) and str(signal.get("action", "HOLD")).upper() == "OPEN":
            trace = signal.get("trace", {}) if isinstance(signal.get("trace", {}), dict) else {}
            evidence = trace.get("evidence", []) if isinstance(trace.get("evidence", []), list) else []
            evidence.append("instance_reduce_only")
            trace["evidence"] = evidence
            signal["trace"] = trace
            signal["action"] = "HOLD"
            signal["side"] = "NONE"
            signal["reject_reason"] = "INSTANCE_REDUCE_ONLY"
        max_positions_limit = int(float(signal["param_overrides"].get("max_positions", 0.0) or 0.0)) if "max_positions" in signal["param_overrides"] else 0
        if (
            max_positions_limit > 0
            and current_open_positions >= max_positions_limit
            and str(signal.get("action", "HOLD")).upper() == "OPEN"
        ):
            trace = signal.get("trace", {}) if isinstance(signal.get("trace", {}), dict) else {}
            evidence = trace.get("evidence", []) if isinstance(trace.get("evidence", []), list) else []
            evidence.append("instance_max_positions_reached")
            trace["evidence"] = evidence
            signal["trace"] = trace
            signal["action"] = "HOLD"
            signal["side"] = "NONE"
            signal["reject_reason"] = f"INSTANCE_MAX_POSITIONS_{current_open_positions}>={max_positions_limit}"
        if (
            allowed_side in {"BUY", "SELL"}
            and str(signal.get("action", "HOLD")).upper() == "OPEN"
            and str(signal.get("side", "NONE") or "NONE").upper() in {"BUY", "SELL"}
            and str(signal.get("side", "NONE") or "NONE").upper() != allowed_side
        ):
            trace = signal.get("trace", {}) if isinstance(signal.get("trace", {}), dict) else {}
            evidence = trace.get("evidence", []) if isinstance(trace.get("evidence", []), list) else []
            evidence.append("route_allowed_side_block")
            trace["evidence"] = evidence
            signal["trace"] = trace
            signal["action"] = "HOLD"
            signal["side"] = "NONE"
            signal["reject_reason"] = f"ROUTE_ALLOWED_SIDE_BLOCK:{allowed_side}"
        return signal

    if entry_selection_active and len(target_instances) == 1:
        primary_signal = _render_instance_signal(target_instances[0])
        signals.append(primary_signal)
        primary_reason = str(primary_signal.get("reject_reason", "") or "")
        if primary_reason.startswith("ROUTE_ALLOWED_SIDE_BLOCK:") and allowed_side in {"BUY", "SELL"}:
            fallback_candidates = [
                instance
                for instance in instances
                if str(instance.get("instance_id", "") or "").strip() != selected_entry_instance_id
            ]
            fallback_signals = [_render_instance_signal(instance) for instance in fallback_candidates]
            actionable_fallbacks = [signal for signal in fallback_signals if _is_actionable_signal(signal)]
            if actionable_fallbacks:
                selected_fallback = max(
                    actionable_fallbacks,
                    key=lambda signal: (
                        abs(_signal_signed_score(signal, float(signal.get("instance_weight", 0.0) or 0.0))),
                        float((signal.get("trace", {}) or {}).get("confidence", 0.0) or 0.0),
                        str(signal.get("_instance_id", "") or ""),
                    ),
                )
                fallback_instance_id = str(selected_fallback.get("_instance_id", "") or "").strip()
                if fallback_instance_id:
                    primary_trace = (
                        primary_signal.get("trace", {})
                        if isinstance(primary_signal.get("trace", {}), dict)
                        else {}
                    )
                    primary_evidence = (
                        primary_trace.get("evidence", [])
                        if isinstance(primary_trace.get("evidence", []), list)
                        else []
                    )
                    if "entry_owner_route_side_failover" not in primary_evidence:
                        primary_evidence.append("entry_owner_route_side_failover")
                    primary_trace["evidence"] = primary_evidence
                    primary_signal["trace"] = primary_trace
                    for fallback_signal in fallback_signals:
                        if str(fallback_signal.get("_instance_id", "") or "").strip() != fallback_instance_id:
                            continue
                        fallback_signal["entry_selected"] = True
                        fallback_signal["_entry_owner_override"] = fallback_instance_id
                        fallback_signal["_entry_failover_reason"] = "ENTRY_OWNER_ROUTE_ALLOWED_SIDE_BLOCK"
                        fallback_trace = (
                            fallback_signal.get("trace", {})
                            if isinstance(fallback_signal.get("trace", {}), dict)
                            else {}
                        )
                        fallback_evidence = (
                            fallback_trace.get("evidence", [])
                            if isinstance(fallback_trace.get("evidence", []), list)
                            else []
                        )
                        if "entry_owner_route_side_failover_selected" not in fallback_evidence:
                            fallback_evidence.append("entry_owner_route_side_failover_selected")
                        fallback_trace["evidence"] = fallback_evidence
                        fallback_signal["trace"] = fallback_trace
                        signals.append(fallback_signal)
                        break
        return signals

    # Parallel execution when multiple instances — each run_engine() is
    # an independent CPU-bound rule evaluation with no shared mutable state.
    if len(target_instances) <= 1:
        for instance in target_instances:
            signals.append(_render_instance_signal(instance))
    else:
        with ThreadPoolExecutor(max_workers=min(len(target_instances), 4)) as pool:
            future_map = {pool.submit(_render_instance_signal, inst): inst for inst in target_instances}
            for future in as_completed(future_map):
                try:
                    signals.append(future.result())
                except Exception as exc:
                    instance = future_map[future]
                    instance_id = str(instance.get("instance_id", "?") or "?")
                    signals.append({
                        "action": "HOLD", "side": "NONE",
                        "reject_reason": f"INSTANCE_CRASH:{instance_id}:{exc}",
                        "_instance_id": instance_id,
                        "_sig_strategy_id": str(instance.get("strategy_id", "") or ""),
                        "instance_weight": 0.0,
                        "entry_selected": False,
                        "trace": {"confidence": 0.0, "evidence": [f"instance_crash:{instance_id}"]},
                    })
    # Preserve original insertion order for deterministic aggregation
    instance_order = {str(inst.get("instance_id", "") or ""): idx for idx, inst in enumerate(target_instances)}
    signals.sort(key=lambda s: instance_order.get(str(s.get("_instance_id", "") or ""), 9999))
    return signals


def _aggregate_instance_signals(
    strategy_signals: list[Dict[str, Any]],
    scheduler_snapshot: Dict[str, Any],
    agent_context: Optional[Dict[str, Any]] = None,
    *,
    project_root: Optional[Path] = None,
) -> tuple[Dict[str, Any], Dict[str, Any]]:
    if not strategy_signals:
        now_iso = datetime.now(timezone.utc).isoformat()
        return (
            {
                "signal_id": f"empty_scheduler_{int(time.time())}",
                "timestamp": now_iso,
                "symbol": "XAUUSD",
                "action": "HOLD",
                "side": "NONE",
                "mode": "UNCERTAIN",
                "governor_status": "REJECTED",
                "reject_reason": "EMPTY_SCHEDULER_SNAPSHOT",
                "trace": {"confidence": 0.0, "evidence": ["empty_scheduler_snapshot"], "risk_flags": ["SCHEDULER_EMPTY"]},
            },
            {
                "action": "HOLD",
                "side": "NONE",
                "net_score": 0.0,
                "confidence": 0.0,
                "contributors": [],
                "blocked_reasons": ["EMPTY_SCHEDULER_SNAPSHOT"],
                "selected_instance_id": "",
                "selected_strategy_id": "",
            },
        )

    contributors: list[Dict[str, Any]] = []
    blocked_reasons: list[str] = []
    selected_signal: Optional[Dict[str, Any]] = None
    selected_executable_signal: Optional[Dict[str, Any]] = None
    scheduler_selected_instance_id = str(scheduler_snapshot.get("selected_instance_id", "") or "").strip()
    route = scheduler_snapshot.get("route", {}) if isinstance(scheduler_snapshot.get("route", {}), dict) else {}
    raw_decision_policy = (
        (agent_context or {}).get("decision_policy", {})
        if isinstance((agent_context or {}).get("decision_policy", {}), dict)
        else {}
    )
    decision_policy = _normalize_decision_policy(raw_decision_policy or scheduler_snapshot.get("decision_policy", {}))
    interpreter_context = (
        (agent_context or {}).get("interpreter", {})
        if isinstance((agent_context or {}).get("interpreter", {}), dict)
        else {}
    )
    execution_constraint = _resolve_execution_constraint(agent_context or {}, route, decision_policy)
    execution_constraint_payload = _execution_constraint_to_dict(execution_constraint)
    policy_entry_policy = str(
        decision_policy.get("execution", {}).get("entry_policy", "")
        if isinstance(decision_policy.get("execution", {}), dict)
        else ""
    ).strip()
    entry_policy = policy_entry_policy or str(route.get("entry_policy", "") or "").strip()
    entry_owner_instance_id = str(route.get("entry_owner_instance_id", scheduler_selected_instance_id) or scheduler_selected_instance_id).strip()
    entry_owner_override = next(
        (
            str(signal.get("_entry_owner_override", "") or "").strip()
            for signal in strategy_signals
            if str(signal.get("_entry_owner_override", "") or "").strip()
        ),
        "",
    )
    entry_failover_reason = next(
        (
            str(signal.get("_entry_failover_reason", "") or "").strip()
            for signal in strategy_signals
            if str(signal.get("_entry_failover_reason", "") or "").strip()
        ),
        "",
    )
    strict_selected_entry = entry_policy == "selected_strategy_only"
    health_weighted_vote = entry_policy == "health_weighted_vote"
    any_actionable = entry_policy == "any_actionable"
    if strict_selected_entry and entry_owner_override:
        entry_owner_instance_id = entry_owner_override

    # ── Health-weighted voting weights ──
    voting_weights: Dict[str, float] = {}
    if health_weighted_vote and project_root is not None:
        hub = get_intelligence_hub()
        instances_for_vote = scheduler_snapshot.get("instances", []) if isinstance(scheduler_snapshot.get("instances", []), list) else []
        vote_health = hub.refresh_all(project_root, instances_for_vote)
        regime_for_vote = str(route.get("regime", "") or "")
        voting_weights = hub.compute_voting_weights(vote_health, current_regime=regime_for_vote)
    selected_score = 0.0
    selected_executable_score = 0.0
    raw_net_score = 0.0
    eligible_weight_total = 0.0
    eligible_count = 0
    executable_weight_total = 0.0
    executable_count = 0
    min_confidence = float(decision_policy.get("execution", {}).get("min_confidence", 0.15) or 0.15)
    for signal in strategy_signals:
        instance_id = str(signal.get("_instance_id", "")).strip()
        strategy_id = str(signal.get("_sig_strategy_id", signal.get("strategy_id", "")) or "").strip()
        action = str(signal.get("action", "HOLD") or "HOLD").upper()
        if strict_selected_entry and action == "OPEN":
            if not entry_owner_instance_id:
                signal["action"] = "HOLD"
                signal["side"] = "NONE"
                signal["reject_reason"] = "ENTRY_SELECTION_MISSING"
            elif instance_id != entry_owner_instance_id:
                signal["action"] = "HOLD"
                signal["side"] = "NONE"
                signal["reject_reason"] = "ENTRY_SELECTION_MISMATCH"
        weight = float(signal.get("instance_weight", 0.0) or 0.0)
        # Health-weighted vote: adjust instance weight by health-derived voting power
        if health_weighted_vote and voting_weights:
            vote_w = voting_weights.get(instance_id, weight)
            if vote_w <= 0.0 and action == "OPEN":
                signal["action"] = "HOLD"
                signal["side"] = "NONE"
                signal["reject_reason"] = "HEALTH_VOTE_ZERO_WEIGHT"
            else:
                weight = vote_w
                signal["instance_weight"] = round(vote_w, 4)
        opportunity = _build_strategy_opportunity(signal, interpreter_context)
        materialized = _materialize_opportunity(opportunity, execution_constraint, min_confidence=min_confidence)
        opportunity_payload = _strategy_opportunity_to_dict(opportunity)
        materialized_payload = _materialized_opportunity_to_dict(materialized)
        signal["opportunity"] = opportunity_payload
        signal["execution_constraint"] = execution_constraint_payload
        signal["materialized_opportunity"] = materialized_payload
        signed_score = (
            _opportunity_signed_score(opportunity, weight)
            if materialized.eligible_for_aggregation
            else 0.0
        )
        confidence = float((signal.get("trace", {}) or {}).get("confidence", 0.0) or 0.0)
        reason = str(signal.get("reject_reason", "") or "")
        contributor = {
            "instance_id": instance_id,
            "strategy_id": strategy_id,
            "action": str(signal.get("action", "HOLD") or "HOLD"),
            "side": str(signal.get("side", "NONE") or "NONE"),
            "weight": round(weight, 4),
            "confidence": round(confidence, 4),
            "signed_score": round(signed_score, 6),
            "reason": reason,
            "directional_view": opportunity.directional_view,
            "setup_quality": opportunity.setup_quality,
            "entry_style_preference": opportunity.entry_style_preference,
            "urgency": opportunity.urgency,
            "materialized_action": materialized.materialized_action,
            "materialized_reason": materialized.reason,
            "eligible_for_aggregation": materialized.eligible_for_aggregation,
        }
        contributors.append(contributor)
        raw_net_score += signed_score
        if reason:
            blocked_reasons.append(f"{instance_id}:{reason}" if instance_id else reason)
        elif materialized.materialized_action == "BLOCKED":
            blocked_reasons.append(f"{instance_id}:{materialized.reason}" if instance_id else materialized.reason)
        if abs(signed_score) >= abs(selected_score):
            selected_signal = signal
            selected_score = signed_score
        if materialized.eligible_for_aggregation:
            eligible_count += 1
            eligible_weight_total += max(0.0, weight)
        if materialized.final_action == "OPEN" and materialized.final_side in {"BUY", "SELL"}:
            executable_count += 1
            executable_weight_total += max(0.0, weight)
            if abs(signed_score) >= abs(selected_executable_score):
                selected_executable_signal = signal
                selected_executable_score = signed_score

    if selected_executable_signal is not None:
        selected_signal = selected_executable_signal
    elif scheduler_selected_instance_id:
        selected_signal = next(
            (
                signal
                for signal in strategy_signals
                if str(signal.get("_instance_id", "") or "").strip() == scheduler_selected_instance_id
            ),
            selected_signal,
        )
    elif selected_signal is None:
        selected_signal = deepcopy(strategy_signals[0])
    aggregated_signal = deepcopy(selected_signal)
    selected_opportunity_payload = (
        aggregated_signal.get("opportunity", {})
        if isinstance(aggregated_signal.get("opportunity", {}), dict)
        else {}
    )

    # ── Cross-strategy correlation evaluation ──
    correlation: Dict[str, Any] = {"consensus": "SINGLE_SIGNAL", "confidence_adjustment": 1.0, "volume_adjustment": 1.0, "cooldown_adjustment_sec": 0, "net_bias": "NONE"}
    risk_config = decision_policy.get("risk", {}) if isinstance(decision_policy, dict) else {}
    corr_config = risk_config.get("cross_strategy", {}).get("correlation_tracking", {}) if isinstance(risk_config.get("cross_strategy", {}), dict) else {}
    if isinstance(corr_config, dict) and corr_config.get("enabled", True) and project_root is not None:
        hub_corr = get_intelligence_hub()
        instances_for_corr = scheduler_snapshot.get("instances", []) if isinstance(scheduler_snapshot.get("instances", []), list) else []
        corr_health = hub_corr.refresh_all(project_root, instances_for_corr)
        correlation = hub_corr.evaluate_cross_strategy_correlation(strategy_signals, corr_health)
        # Apply correlation adjustments
        if correlation["consensus"] == "DIVERGENT":
            divergence_penalty = float(corr_config.get("divergence_confidence_penalty", 0.4) or 0.4)
            divergence_vol_penalty = float(corr_config.get("divergence_volume_penalty", 0.6) or 0.6)
            correlation["confidence_adjustment"] = max(0.5, 1.0 - divergence_penalty)
            correlation["volume_adjustment"] = max(0.4, 1.0 - divergence_vol_penalty)
            correlation["cooldown_adjustment_sec"] = int(corr_config.get("divergence_cooldown_sec", 120) or 120)
        elif correlation["consensus"] in {"ALIGNED_BUY", "ALIGNED_SELL"}:
            alignment_bonus = float(corr_config.get("alignment_volume_bonus", 0.05) or 0.05)
            correlation["confidence_adjustment"] = 1.0
            correlation["volume_adjustment"] = min(1.15, 1.0 + alignment_bonus * correlation.get("aligned_instances", []).__len__())
            correlation["cooldown_adjustment_sec"] = -30

    effective_net_score = raw_net_score
    if eligible_weight_total > 0.0:
        effective_net_score = raw_net_score / eligible_weight_total
    # Apply correlation confidence adjustment
    effective_net_score *= float(correlation.get("confidence_adjustment", 1.0) or 1.0)
    effective_net_score = max(-0.99, min(0.99, effective_net_score))
    aggregated_confidence = round(abs(effective_net_score), 4)
    net_side = "BUY" if effective_net_score > 0 else "SELL" if effective_net_score < 0 else "NONE"
    hold_reason = ""
    eligible_weight_floor = min(0.18, max(0.05, min_confidence * 0.8))
    if strict_selected_entry and not entry_owner_instance_id:
        aggregated_signal["action"] = "HOLD"
        aggregated_signal["side"] = "NONE"
        hold_reason = "AGGREGATED_HOLD_NO_SELECTED_ENTRY"
        aggregated_signal["reject_reason"] = hold_reason
    elif eligible_count == 0:
        aggregated_signal["action"] = "HOLD"
        aggregated_signal["side"] = "NONE"
        hold_reason = "AGGREGATED_HOLD_NO_ELIGIBLE_OPPORTUNITY"
        aggregated_signal["reject_reason"] = hold_reason
    elif eligible_weight_total < eligible_weight_floor:
        aggregated_signal["action"] = "HOLD"
        aggregated_signal["side"] = "NONE"
        hold_reason = "AGGREGATED_HOLD_LOW_ELIGIBLE_WEIGHT"
        aggregated_signal["reject_reason"] = hold_reason
    elif aggregated_confidence < min_confidence or net_side == "NONE":
        aggregated_signal["action"] = "HOLD"
        aggregated_signal["side"] = "NONE"
        hold_reason = "AGGREGATED_HOLD_LOW_NET_SCORE"
        aggregated_signal["reject_reason"] = hold_reason
    aggregated_opportunity = StrategyOpportunity(
        strategy_id=str(aggregated_signal.get("_sig_strategy_id", aggregated_signal.get("strategy_id", "")) or ""),
        instance_id=str(aggregated_signal.get("_instance_id", aggregated_signal.get("instance_id", "")) or ""),
        directional_view=net_side if net_side in {"BUY", "SELL"} else "NONE",
        setup_quality=_opportunity_setup_quality("OPEN", net_side, aggregated_confidence) if net_side in {"BUY", "SELL"} else "INVALID",
        entry_style_preference=str(selected_opportunity_payload.get("entry_style_preference", "MARKET") or "MARKET"),
        long_opportunity_score=aggregated_confidence if net_side == "BUY" else 0.0,
        short_opportunity_score=aggregated_confidence if net_side == "SELL" else 0.0,
        urgency=str(selected_opportunity_payload.get("urgency", "LOW") or "LOW"),
        confidence=aggregated_confidence,
        actionable=eligible_count > 0 and net_side in {"BUY", "SELL"},
        source_action="OPEN" if eligible_count > 0 else "HOLD",
        source_side=net_side,
        notes=[f"eligible_count={eligible_count}", f"executable_count={executable_count}"],
    )
    aggregated_materialized = _materialize_opportunity(
        aggregated_opportunity,
        execution_constraint,
        min_confidence=min_confidence,
    )
    if not hold_reason:
        aggregated_signal["action"] = aggregated_materialized.final_action
        aggregated_signal["side"] = aggregated_materialized.final_side
        aggregated_signal["entry_style"] = aggregated_materialized.entry_style
        aggregated_signal["entry_size_scale"] = round(aggregated_materialized.size_scale, 4)
        aggregated_signal["materialized_action"] = aggregated_materialized.materialized_action
        aggregated_signal["materialized_reason"] = aggregated_materialized.reason
        if aggregated_materialized.final_action == "HOLD" and aggregated_materialized.reason:
            aggregated_signal["reject_reason"] = aggregated_materialized.reason
    else:
        aggregated_signal["materialized_action"] = "BLOCKED"
        aggregated_signal["materialized_reason"] = hold_reason
    base_signal_id = str(aggregated_signal.get("_base_signal_id", aggregated_signal.get("signal_id", f"sig_{int(time.time())}")) or f"sig_{int(time.time())}")
    if len(strategy_signals) > 1:
        fingerprint = hashlib.sha1("|".join(sorted(str(item.get("_instance_id", "")) for item in strategy_signals)).encode("utf-8")).hexdigest()[:8]
        aggregated_signal["signal_id"] = f"{base_signal_id}__agg_{fingerprint}"
    aggregation_payload = {
        "action": str(aggregated_signal.get("action", "HOLD") or "HOLD"),
        "side": str(aggregated_signal.get("side", "NONE") or "NONE"),
        "materialized_action": str(aggregated_signal.get("materialized_action", "") or ""),
        "materialized_reason": str(aggregated_signal.get("materialized_reason", "") or ""),
        "net_score": round(effective_net_score, 6),
        "raw_net_score": round(raw_net_score, 6),
        "confidence": aggregated_confidence,
        "contributors": contributors,
        "blocked_reasons": blocked_reasons,
        "selected_instance_id": str(aggregated_signal.get("_instance_id", "") or ""),
        "selected_strategy_id": str(aggregated_signal.get("_sig_strategy_id", aggregated_signal.get("strategy_id", "")) or ""),
        "selected_reason": str(aggregated_signal.get("reject_reason", "") or ""),
        "entry_policy": entry_policy,
        "entry_owner_instance_id": entry_owner_instance_id,
        "entry_failover_reason": entry_failover_reason,
        "eligible_count": eligible_count,
        "eligible_weight_total": round(eligible_weight_total, 6),
        "eligible_weight_floor": round(eligible_weight_floor, 6),
        "executable_count": executable_count,
        "executable_weight_total": round(executable_weight_total, 6),
        "voting_weights": {iid: round(w, 4) for iid, w in voting_weights.items()} if voting_weights else {},
        "execution_constraint": execution_constraint_payload,
        "selected_opportunity": selected_opportunity_payload,
        "aggregated_opportunity": _strategy_opportunity_to_dict(aggregated_opportunity),
        "aggregated_materialized_opportunity": _materialized_opportunity_to_dict(aggregated_materialized),
        "correlation": {
            "consensus": str(correlation.get("consensus", "UNKNOWN") or "UNKNOWN"),
            "net_bias": str(correlation.get("net_bias", "NONE") or "NONE"),
            "confidence_adjustment": round(float(correlation.get("confidence_adjustment", 1.0) or 1.0), 4),
            "volume_adjustment": round(float(correlation.get("volume_adjustment", 1.0) or 1.0), 4),
            "cooldown_adjustment_sec": int(correlation.get("cooldown_adjustment_sec", 0) or 0),
            "aligned_instances": correlation.get("aligned_instances", []),
            "divergent_instances": correlation.get("divergent_instances", []),
        },
        "per_instance_health": {
            iid: {
                "score": round(h.score, 1),
                "grade": h.grade,
                "suggested_action": h.suggested_action,
                "suggested_scale": round(h.suggested_scale, 4),
                "loss_streak": h.loss_streak,
                "best_regime": h.best_regime,
                "current_regime_fitness": round(h.current_regime_fitness, 4),
                "primary_loss_pattern": h.primary_loss_pattern,
            }
            for iid, h in (get_intelligence_hub().refresh_all(project_root, scheduler_snapshot.get("instances", []) if isinstance(scheduler_snapshot.get("instances", []), list) else [])).items()
        } if project_root is not None else {},
        "scheduler_mode": str(scheduler_snapshot.get("mode", "single_active") or "single_active"),
        "scheduler_reason": str(scheduler_snapshot.get("reason", "runtime_scheduler") or "runtime_scheduler"),
        "decision_policy": decision_policy,
        "agent_context": {
            "macro_direction": str(
                ((agent_context or {}).get("macro", {}) if isinstance((agent_context or {}).get("macro", {}), dict) else {}).get("direction", "NEUTRAL")
                or "NEUTRAL"
            ),
            "effective_macro_direction": str(
                ((agent_context or {}).get("macro", {}) if isinstance((agent_context or {}).get("macro", {}), dict) else {}).get("effective_direction", "NEUTRAL")
                or "NEUTRAL"
            ),
            "binding_macro_direction": _resolve_binding_direction(
                ((agent_context or {}).get("macro", {}) if isinstance((agent_context or {}).get("macro", {}), dict) else {})
            ),
            "execution_advice": str(
                ((agent_context or {}).get("interpreter", {}) if isinstance((agent_context or {}).get("interpreter", {}), dict) else {}).get("execution_advice", "ALLOW_FOLLOW")
                or "ALLOW_FOLLOW"
            ),
            "structural_side": str(
                ((agent_context or {}).get("interpreter", {}) if isinstance((agent_context or {}).get("interpreter", {}), dict) else {}).get("structural_side", "")
                or ""
            ),
            "macro_state": str(
                ((agent_context or {}).get("macro", {}) if isinstance((agent_context or {}).get("macro", {}), dict) else {}).get("state", "UNKNOWN")
                or "UNKNOWN"
            ),
            "macro_bias_weight": float(
                ((agent_context or {}).get("macro", {}) if isinstance((agent_context or {}).get("macro", {}), dict) else {}).get("bias_weight", 1.0)
                or 1.0
            ),
            "route": (
                (agent_context or {}).get("route", {})
                if isinstance((agent_context or {}).get("route", {}), dict)
                else {}
            ),
        },
    }
    if hold_reason and hold_reason not in aggregation_payload["blocked_reasons"]:
        aggregation_payload["blocked_reasons"].append(hold_reason)
    trace = aggregated_signal.get("trace", {}) if isinstance(aggregated_signal.get("trace", {}), dict) else {}
    trace["aggregation"] = {
        "net_score": aggregation_payload["net_score"],
        "raw_net_score": aggregation_payload["raw_net_score"],
        "eligible_count": aggregation_payload["eligible_count"],
        "eligible_weight_total": aggregation_payload["eligible_weight_total"],
        "eligible_weight_floor": aggregation_payload["eligible_weight_floor"],
        "materialized_action": aggregation_payload["materialized_action"],
        "contributors": contributors,
    }
    trace["confidence"] = aggregated_confidence
    aggregated_signal["trace"] = trace
    aggregated_signal["execution_constraint"] = execution_constraint_payload
    aggregated_signal["opportunity"] = aggregation_payload["aggregated_opportunity"]
    aggregated_signal["materialized_opportunity"] = aggregation_payload["aggregated_materialized_opportunity"]
    aggregated_signal["_selected_instance_id"] = aggregation_payload["selected_instance_id"]
    aggregated_signal["_selected_strategy"] = aggregation_payload["selected_strategy_id"]
    return aggregated_signal, aggregation_payload


def _signal_attribution_metadata(
    signal: Dict[str, Any],
    aggregated_intent: Dict[str, Any],
) -> Dict[str, Any]:
    selected_instance_id = str(
        signal.get("_selected_instance_id")
        or aggregated_intent.get("selected_instance_id")
        or signal.get("_instance_id")
        or signal.get("instance_id")
        or ""
    ).strip()
    selected_strategy_id = str(
        signal.get("_selected_strategy")
        or aggregated_intent.get("selected_strategy_id")
        or signal.get("_sig_strategy_id")
        or signal.get("strategy_id")
        or ""
    ).strip()
    contributors = (
        aggregated_intent.get("contributors", [])
        if isinstance(aggregated_intent.get("contributors", []), list)
        else []
    )
    metadata = {
        "action": str(signal.get("action", "HOLD") or "HOLD").upper(),
        "side": str(signal.get("side", "NONE") or "NONE").upper(),
        "symbol": str(signal.get("symbol", "XAUUSD") or "XAUUSD"),
        "entry_style": str(signal.get("entry_style", "MARKET") or "MARKET").upper(),
        "materialized_action": str(signal.get("materialized_action", "") or ""),
        "volume_lots": float(signal.get("volume_lots", signal.get("volume", signal.get("lots", 0.0))) or 0.0),
        "requested_volume_lots": float(signal.get("requested_volume_lots", signal.get("volume_lots", signal.get("volume", signal.get("lots", 0.0)))) or 0.0),
        "configured_base_lots": float(signal.get("configured_base_lots", 0.0) or 0.0),
        "rule_cap_lots": float(signal.get("rule_cap_lots", 0.0) or 0.0),
        "rule_cap_reasons": signal.get("rule_cap_reasons", []) if isinstance(signal.get("rule_cap_reasons", []), list) else [],
        "decision_id": str(signal.get("decision_id", signal.get("signal_id", "")) or ""),
        "trace_id": str(signal.get("trace_id", signal.get("signal_id", "")) or ""),
        "strategy_id": str(signal.get("strategy_id", "") or ""),
        "instance_id": str(signal.get("instance_id", selected_instance_id) or ""),
        "attributed_strategy_id": selected_strategy_id,
        "attributed_instance_id": selected_instance_id,
        "selected_strategy_id": selected_strategy_id,
        "selected_instance_id": selected_instance_id,
        "contributors": contributors,
    }
    if metadata["action"] == "HOLD":
        metadata["hold_diagnostics"] = _hold_diagnostics_summary(aggregated_intent)
    metadata.update(normalize_position_attribution(metadata, action=metadata["action"]))
    return metadata


def _attach_signal_attribution(
    signal: Dict[str, Any],
    aggregated_intent: Dict[str, Any],
) -> Dict[str, Any]:
    metadata = _signal_attribution_metadata(signal, aggregated_intent)
    signal.update(
        {
            "instance_id": metadata["instance_id"],
            "attributed_instance_id": metadata["attributed_instance_id"],
            "attributed_strategy_id": metadata["attributed_strategy_id"],
            "selected_instance_id": metadata["selected_instance_id"],
            "selected_strategy_id": metadata["selected_strategy_id"],
            "decision_id": metadata["decision_id"],
            "trace_id": metadata["trace_id"],
            "contributors": metadata["contributors"],
        }
    )
    return metadata


def _append_instance_execution_event(
    project_root: Path,
    signal: Dict[str, Any],
    report: Optional[Dict[str, Any]],
) -> None:
    action = str(signal.get("action", "HOLD") or "HOLD").upper()
    if action not in {"OPEN", "CLOSE"}:
        return
    payload = report if isinstance(report, dict) else {}
    status = str(payload.get("status", "UNKNOWN") or "UNKNOWN").upper()
    event = {
        "ts": str(payload.get("execution_ts") or payload.get("ts") or datetime.now(timezone.utc).isoformat()),
        "signal_id": str(signal.get("signal_id", "") or ""),
        "decision_id": str(signal.get("decision_id", signal.get("signal_id", "")) or ""),
        "intent_id": str(payload.get("intent_id", signal.get("intent_id", "")) or ""),
        "action": action,
        "side": str(signal.get("side", "NONE") or "NONE").upper(),
        "status": status,
        "reason": str(payload.get("reason", "") or ""),
        "symbol": str(signal.get("symbol", "XAUUSD") or "XAUUSD"),
        "strategy_id": str(signal.get("attributed_strategy_id", signal.get("strategy_id", "")) or ""),
        "instance_id": str(signal.get("attributed_instance_id", signal.get("instance_id", "")) or ""),
        "selected_strategy_id": str(signal.get("selected_strategy_id", signal.get("strategy_id", "")) or ""),
        "selected_instance_id": str(signal.get("selected_instance_id", signal.get("instance_id", "")) or ""),
        "contributors": signal.get("contributors", []) if isinstance(signal.get("contributors", []), list) else [],
        "entry_style": str(signal.get("entry_style", "MARKET") or "MARKET").upper(),
        "materialized_action": str(signal.get("materialized_action", "") or ""),
        "order_id": payload.get("order_id"),
        "deal_id": payload.get("deal_id"),
        "retcode": payload.get("retcode"),
        "reconciliation_ok": payload.get("reconciliation_ok"),
        "volume": float(signal.get("volume_lots", signal.get("volume", signal.get("lots", 0.0))) or 0.0),
        "volume_lots": float(signal.get("volume_lots", signal.get("volume", signal.get("lots", 0.0))) or 0.0),
        "requested_volume_lots": float(signal.get("requested_volume_lots", signal.get("volume_lots", signal.get("volume", signal.get("lots", 0.0)))) or 0.0),
        "configured_base_lots": float(signal.get("configured_base_lots", 0.0) or 0.0),
        "rule_cap_lots": float(signal.get("rule_cap_lots", 0.0) or 0.0),
        "rule_cap_reasons": signal.get("rule_cap_reasons", []) if isinstance(signal.get("rule_cap_reasons", []), list) else [],
    }
    ledger_path = project_root / "logs" / "trading_ledger.ndjson"
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with ledger_path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(event, ensure_ascii=False) + "\n")


def _apply_live_volume_plan(
    signal: Dict[str, Any],
    *,
    project_root: Path,
    aggregated_intent: Optional[Dict[str, Any]] = None,
) -> None:
    action = str(signal.get("action", "HOLD") or "HOLD").upper()
    if action != "OPEN":
        return
    requested_volume = signal.get(
        "requested_volume_lots",
        signal.get("volume_lots", signal.get("volume", signal.get("lots"))),
    )
    volume_plan = resolve_live_volume_plan(requested_volume, project_root=project_root)
    signal["requested_volume_lots"] = volume_plan["requested_volume_lots"]
    signal["configured_base_lots"] = volume_plan["configured_base_lots"]
    signal["rule_cap_lots"] = volume_plan["rule_cap_lots"]
    signal["rule_cap_reasons"] = list(volume_plan["rule_cap_reasons"])
    signal["final_volume_lots"] = volume_plan["final_volume_lots"]
    signal["volume_lots"] = volume_plan["final_volume_lots"]
    entry_size_scale = float(signal.get("entry_size_scale", 1.0) or 1.0)
    if entry_size_scale != 1.0 and float(signal.get("volume_lots", 0) or 0) > 0:
        signal["volume_lots"] = round(float(signal["volume_lots"]) * entry_size_scale, 2)
        signal["entry_size_scale_applied"] = round(entry_size_scale, 4)

    # ── Per-instance health-based scaling ──
    agg = aggregated_intent if isinstance(aggregated_intent, dict) else {}
    per_instance_health = agg.get("per_instance_health", {}) if isinstance(agg.get("per_instance_health", {}), dict) else {}
    instance_id = str(signal.get("_instance_id", signal.get("instance_id", "")) or "").strip()
    health_info = per_instance_health.get(instance_id, {}) if isinstance(per_instance_health.get(instance_id, {}), dict) else {}
    health_scale = float(health_info.get("suggested_scale", 1.0) or 1.0)
    if health_scale < 1.0 and float(signal.get("volume_lots", 0) or 0) > 0:
        signal["volume_lots"] = round(float(signal["volume_lots"]) * health_scale, 2)
        signal["health_scale_applied"] = round(health_scale, 4)
        if health_scale <= 0.0:
            signal["action"] = "HOLD"
            signal["side"] = "NONE"
            signal["reject_reason"] = "HEALTH_SHADOW_ONLY"
            return

    # ── Cross-strategy correlation volume adjustment ──
    correlation = agg.get("correlation", {}) if isinstance(agg.get("correlation", {}), dict) else {}
    corr_vol_adj = float(correlation.get("volume_adjustment", 1.0) or 1.0)
    if corr_vol_adj != 1.0 and float(signal.get("volume_lots", 0) or 0) > 0:
        signal["volume_lots"] = round(float(signal["volume_lots"]) * corr_vol_adj, 2)
        signal["correlation_volume_applied"] = round(corr_vol_adj, 4)

    # ── Risk budget: cap per-instance volume ──
    risk_config = agg.get("decision_policy", {}).get("risk", {}) if isinstance(agg.get("decision_policy", {}), dict) else {}
    budget_config = risk_config.get("risk_budget", {}) if isinstance(risk_config, dict) else {}
    max_lot_per = float(budget_config.get("max_lot_per_instance", 0.20) or 0.20)
    if float(signal.get("volume_lots", 0) or 0) > max_lot_per:
        signal["volume_lots"] = round(max_lot_per, 2)
        signal["risk_budget_cap_applied"] = round(max_lot_per, 4)

    # ── Interpreter directional position multiplier ──
    bias_mult = float(signal.get("interpreter_position_multiplier", 1.0) or 1.0)
    if bias_mult != 1.0 and float(signal.get("volume_lots", 0) or 0) > 0:
        signal["volume_lots"] = round(float(signal["volume_lots"]) * bias_mult, 2)
        signal["interpreter_bias_applied"] = round(bias_mult, 4)


def _evaluate_runtime_prechecks(
    ctx: PipelineRunContext,
    *,
    runtime_state: Dict[str, Any],
) -> Optional[PipelineOutcome]:
    if bool(runtime_state.get("kill_switch", False)):
        ctx.signal = _build_idle_signal(
            signal_id=f"kill_switch_{int(time.time())}",
            symbol=ctx.symbol,
            reject_reason="KILL_SWITCH_ACTIVE",
            evidence=["kill_switch_active"],
            risk_flags=["KILL_SWITCH"],
        )
        return PipelineOutcome(
            ledger_events=["kill_switch"],
            include_runtime=False,
        )

    # ── 浮动亏损熔断：浮亏超过余额 5% 时阻止开新仓 ──
    floating_check = _check_floating_loss_limits(ctx)
    if floating_check is not None:
        return floating_check

    exec_stats = aggregate_exec_stats(ctx.project_root / "logs")
    gate_check = check_rollout_safety(ctx.project_root, ctx.tenant_id, exec_stats, runtime_state)
    if not gate_check.get("safe", True):
        reason = str(gate_check.get("reason", "ROLLBACK_UNSAFE") or "ROLLBACK_UNSAFE")
        ctx.signal = _build_idle_signal(
            signal_id=f"gate_blocked_{int(time.time())}",
            symbol=ctx.symbol,
            reject_reason=reason,
            evidence=[f"gate_check_failed:{reason}"],
            risk_flags=["ROLLBACK_GATE", reason],
        )
        ctx.promotion_info = {
            "blocked": True,
            "reason": reason,
            "trigger": gate_check.get("trigger"),
            "rollback": gate_check.get("rollback"),
        }
        return PipelineOutcome(
            ledger_events=["gate_blocked"],
            include_runtime=False,
        )

    ctx.promotion_info = {"mode": gate_check.get("promotion_mode", "shadow")}
    ctx.promotion_info["observe_only"] = ctx.promotion_info.get("mode") == "shadow"
    gate_readiness = evaluate_promotion_gates(ctx.project_root, ctx.tenant_id, exec_stats)
    if gate_readiness:
        ctx.promotion_info["gate_readiness"] = gate_readiness
    return None


def _check_floating_loss_limits(ctx: PipelineRunContext) -> Optional[PipelineOutcome]:
    """浮动亏损熔断：已持仓时浮亏超过余额阈值则阻止新开仓。

    仅作用于 OPEN 意图 — 允许平仓和 HOLD 不受限制。
    同时计算 floating_pnl_pct 供下游风控消费。
    """
    account = ctx.engine_request.get("account_metrics", {}) if isinstance(ctx.engine_request.get("account_metrics", {}), dict) else {}
    floating_pnl = float(account.get("floating_pnl", 0.0) or 0.0)
    balance = float(account.get("balance", 0.0) or account.get("equity", 0.0) or 0.0)
    position_open = bool(account.get("position_open", False))

    if balance > 0:
        floating_pnl_pct = round(floating_pnl / balance * 100.0, 2)
        account["floating_pnl_pct"] = floating_pnl_pct
    else:
        floating_pnl_pct = 0.0
        account["floating_pnl_pct"] = 0.0

    # 只对开仓做限制
    if not position_open or floating_pnl >= 0:
        return None

    # 浮亏超过 5% 余额 → 熔断
    if floating_pnl_pct <= -5.0:
        ctx.signal = _build_idle_signal(
            signal_id=f"floating_loss_circuit_{int(time.time())}",
            symbol=ctx.symbol,
            reject_reason="FLOATING_LOSS_CIRCUIT_BREAKER",
            evidence=[f"floating_pnl_pct:{floating_pnl_pct:.2f}%"],
            risk_flags=["FLOATING_LOSS_CIRCUIT"],
        )
        return PipelineOutcome(
            ledger_events=["floating_loss_blocked"],
            include_runtime=False,
        )

    return None


def _load_run_context(
    ctx: PipelineRunContext,
    *,
    input_path: Path,
    live_client: Optional[Any],
) -> None:
    ctx.mt5_snapshot = _load_snapshot(ctx.source, input_path, live_client)
    ctx.engine_request = to_engine_request(ctx.mt5_snapshot)
    _inject_account_risk_metrics(ctx.engine_request, ctx.project_root)
    ctx.engine_request["runtime_feedback"] = ctx.mt5_snapshot.get("runtime_feedback", {})
    ctx.scheduler_snapshot = _load_scheduler_snapshot(ctx.project_root, ctx.engine_request)
    ctx.agent_context = _load_agent_decision_context(ctx.project_root, ctx.scheduler_snapshot)
    ctx.scheduler_snapshot = _route_scheduler_snapshot(ctx.scheduler_snapshot, ctx.agent_context, ctx.engine_request)
    ctx.agent_context["route"] = (
        ctx.scheduler_snapshot.get("route", {})
        if isinstance(ctx.scheduler_snapshot.get("route", {}), dict)
        else {}
    )
    ctx.strategy_signals = _build_instance_signals(ctx.engine_request, ctx.project_root, ctx.scheduler_snapshot)
    ctx.strategy_signals = _apply_agent_decision_policy(ctx.strategy_signals, ctx.agent_context)
    ctx.signal, ctx.aggregated_intent = _aggregate_instance_signals(
        ctx.strategy_signals,
        ctx.scheduler_snapshot,
        ctx.agent_context,
        project_root=ctx.project_root,
    )
    ctx.signal_attribution = _attach_signal_attribution(ctx.signal, ctx.aggregated_intent)
    _sync_multi_strategy_blackboard(
        ctx.project_root,
        ctx.scheduler_snapshot,
        ctx.strategy_signals,
        ctx.aggregated_intent,
    )
    _emit_scheduler_health_alerts(ctx.project_root, ctx.scheduler_snapshot)
    ctx.signal_id = str(ctx.signal.get("signal_id"))
    ctx.state_path = ctx.project_root / "data/execution_state.json"


def _evaluate_market_data_delay_gate(ctx: PipelineRunContext) -> Optional[PipelineOutcome]:
    market_metrics = (
        ctx.engine_request.get("market_metrics", {})
        if isinstance(ctx.engine_request.get("market_metrics", {}), dict)
        else {}
    )
    delay_ms = float(market_metrics.get("market_data_delay_ms", 0.0) or 0.0)
    threshold_ms = _resolve_market_data_delay_gate_ms(ctx.project_root, ctx.runtime_state)
    if threshold_ms <= 0 or delay_ms <= threshold_ms:
        return None
    ctx.signal = _build_idle_signal(
        signal_id=f"market_data_delay_gate_{int(time.time())}",
        symbol=ctx.symbol,
        reject_reason="MARKET_DATA_DELAY_GATE",
        evidence=[
            f"market_data_delay_ms={delay_ms:.0f}",
            f"market_data_delay_threshold_ms={threshold_ms:.0f}",
        ],
        risk_flags=["MARKET_DATA_DELAY_GATE", "MARKET_DATA_STALE"],
    )
    return PipelineOutcome(
        execution_report=_build_execution_report(
            str(ctx.signal.get("signal_id", "")),
            "REJECTED",
            "MARKET_DATA_DELAY_GATE",
            market_data_delay_ms=round(delay_ms, 3),
            market_data_delay_threshold_ms=round(threshold_ms, 3),
        ),
        ledger_events=["market_data_delay_gate"],
        append_runtime_log=True,
    )


def _evaluate_retry_outcome(ctx: PipelineRunContext) -> Optional[PipelineOutcome]:
    if ctx.state_path is None:
        raise RuntimeError("pipeline state_path is not initialized")
    action_upper = str(ctx.signal.get("action", "HOLD") or "HOLD").upper()
    if action_upper == "HOLD":
        return None
    create_or_get_signal_state(ctx.state_path, ctx.signal_id)
    existing = get_signal_state(ctx.state_path, ctx.signal_id) or {}
    retry_meta = _state_retry_meta(existing)
    now_epoch = time.time()
    last_epoch = _ts_to_epoch(str(retry_meta.get("last_ts", "")))
    age = max(0.0, now_epoch - last_epoch) if last_epoch > 0 else 10**9
    if is_final_state(ctx.state_path, ctx.signal_id):
        return PipelineOutcome(
            execution_report=_build_execution_report(
                ctx.signal_id,
                "ACKED",
                f"DUPLICATE_SIGNAL_BLOCKED_STATE_{existing.get('state', 'UNKNOWN')}",
            ),
            ledger_events=["duplicate_signal_blocked"],
            append_runtime_log=True,
        )
    if retry_meta.get("state") == "ACKED" and age < ACK_RETRY_COOLDOWN_SECONDS:
        return PipelineOutcome(
            execution_report=_build_execution_report(ctx.signal_id, "ACKED", "ACK_RETRY_THROTTLED"),
            ledger_events=["ack_retry_throttled"],
            append_runtime_log=True,
        )
    if retry_meta.get("state") == "TIMEOUT":
        timeout_count = int(retry_meta.get("timeout_count", 0))
        if timeout_count >= MAX_TIMEOUT_RETRIES:
            transition_state(
                ctx.state_path,
                ctx.signal_id,
                "REJECTED",
                reason="RETRY_EXHAUSTED_TIMEOUT",
                metadata=ctx.signal_attribution,
            )
            return PipelineOutcome(
                execution_report=_build_execution_report(ctx.signal_id, "REJECTED", "RETRY_EXHAUSTED_TIMEOUT"),
                ledger_events=["timeout_retry_exhausted"],
                append_runtime_log=True,
            )
        if age < TIMEOUT_RETRY_COOLDOWN_SECONDS:
            return PipelineOutcome(
                execution_report=_build_execution_report(ctx.signal_id, "TIMEOUT", "TIMEOUT_RETRY_THROTTLED"),
                ledger_events=["timeout_retry_throttled"],
                append_runtime_log=True,
            )
    return None


def _evaluate_guard_outcome(ctx: PipelineRunContext) -> Optional[PipelineOutcome]:
    if ctx.state_path is None:
        ctx.state_path = ctx.project_root / "data/execution_state.json"
    macro_context = (
        ctx.agent_context.get("macro", {}) if isinstance(ctx.agent_context.get("macro", {}), dict) else {}
    )
    macro_authority = _resolve_macro_authority_context(macro_context)
    nominal_binding_direction = _resolve_binding_direction(macro_context)
    binding_direction = _resolve_execution_binding_direction(macro_context)
    if binding_direction == "NEUTRAL":
        agg_macro = (
            ctx.aggregated_intent.get("agent_context", {})
            if isinstance(ctx.aggregated_intent.get("agent_context", {}), dict)
            else {}
        )
        fallback_binding = str(agg_macro.get("binding_macro_direction", "NEUTRAL") or "NEUTRAL").upper()
        fallback_authority = str(agg_macro.get("macro_authority", "DISABLED") or "DISABLED").upper()
        if fallback_authority == "HARD_BIND" and fallback_binding in {"LONG", "SHORT"}:
            binding_direction = fallback_binding
    action_upper = str(ctx.signal.get("action", "HOLD") or "HOLD").upper()
    signal_side_upper = str(ctx.signal.get("side", "NONE") or "NONE").upper()
    if (
        action_upper == "OPEN"
        and signal_side_upper in {"BUY", "SELL"}
        and binding_direction in {"LONG", "SHORT"}
        and _binding_side_conflicts(binding_direction, signal_side_upper)
    ):
        block_reason = "MACRO_DIRECTION_BINDING_BLOCK"
        transition_state(ctx.state_path, ctx.signal_id, "REJECTED", reason=block_reason, metadata=ctx.signal_attribution)
        return _build_binding_rejection_outcome(
            ctx,
            binding_direction=nominal_binding_direction,
            block_reason=block_reason,
        )

    structural_guard = (
        (
            (
                ctx.agent_context.get("interpreter", {})
                if isinstance(ctx.agent_context.get("interpreter", {}), dict)
                else {}
            ).get("hard_conflict_guard", {})
        )
        if isinstance(ctx.agent_context, dict)
        else {}
    )
    if (
        action_upper == "OPEN"
        and signal_side_upper in {"BUY", "SELL"}
        and isinstance(structural_guard, dict)
        and bool(structural_guard.get("enabled", False))
        and macro_authority != "HARD_BIND"
        and _structural_side_conflicts(
            str(structural_guard.get("structural_side", "NONE") or "NONE").upper(),
            signal_side_upper,
        )
    ):
        block_reason = "STRUCTURAL_DIRECTION_CONFLICT_BLOCK"
        ctx.signal["structural_conflict_guard"] = dict(structural_guard)
        ctx.aggregated_intent["structural_conflict_guard"] = dict(structural_guard)
        transition_state(ctx.state_path, ctx.signal_id, "REJECTED", reason=block_reason, metadata=ctx.signal_attribution)
        return _build_binding_rejection_outcome(
            ctx,
            binding_direction=nominal_binding_direction,
            block_reason=block_reason,
        )

    if action_upper == "OPEN" and signal_side_upper in {"BUY", "SELL"}:
        incident_controls = _incident_controls(ctx.runtime_state)
        if bool(incident_controls.get("disable_new_open", False)):
            block_reason = "DISABLE_NEW_OPEN_ACTIVE"
            transition_state(ctx.state_path, ctx.signal_id, "REJECTED", reason=block_reason, metadata=ctx.signal_attribution)
            return _build_open_guard_rejection_outcome(
                ctx,
                block_reason=block_reason,
                guard_snapshot={
                    "allow_open": False,
                    "reason": block_reason,
                    "disable_new_open": True,
                    "disable_new_open_reason": str(incident_controls.get("disable_new_open_reason", "") or "incident_control"),
                    "runtime_mode": str(incident_controls.get("runtime_mode", "") or ""),
                },
            )
        open_guard = resolve_live_open_guard(
            project_root=ctx.project_root,
            side=signal_side_upper,
        )
        if not bool(open_guard.get("allow_open", True)):
            block_reason = str(open_guard.get("reason", "") or "RUNTIME_OPEN_GUARD_BLOCK")
            transition_state(ctx.state_path, ctx.signal_id, "REJECTED", reason=block_reason, metadata=ctx.signal_attribution)
            return _build_open_guard_rejection_outcome(
                ctx,
                block_reason=block_reason,
                guard_snapshot=open_guard,
            )
    return None


def _sync_request_side_effects(ctx: PipelineRunContext) -> None:
    action_upper = str(ctx.signal.get("action", "HOLD") or "HOLD").upper()
    if action_upper == "OPEN":
        sync_open_request(ctx.signal_attribution, data_dir=ctx.project_root / "data", reason="pipeline_dispatch")
        return
    if action_upper in {"CLOSE", "REDUCE"}:
        sync_close_request(
            ctx.signal_attribution,
            data_dir=ctx.project_root / "data",
            ticket=int(ctx.signal.get("ticket", 0) or 0),
            reason="pipeline_dispatch",
        )


def _apply_execution_side_effects(
    ctx: PipelineRunContext,
    report: Optional[Dict[str, Any]],
) -> Optional[Dict[str, Any]]:
    if report is None:
        return None
    if ctx.state_path is None:
        raise RuntimeError("pipeline state_path is not initialized")
    report.update(ctx.signal_attribution)
    action_upper = str(ctx.signal.get("action", "HOLD") or "HOLD").upper()
    if str(report.get("status", "")).upper() == "FILLED":
        if action_upper == "OPEN":
            confirmed_ticket = int(
                report.get("deal_id", 0)
                or report.get("ticket", 0)
                or ctx.signal.get("ticket", 0)
                or 0
            )
            if confirmed_ticket > 0:
                sync_open_fill(
                    ctx.signal_attribution,
                    data_dir=ctx.project_root / "data",
                    ticket=confirmed_ticket,
                    side=str(ctx.signal.get("side", "NONE") or "NONE"),
                    volume=float(ctx.signal.get("volume_lots", 0.0) or 0.0),
                    entry_price=float(ctx.signal.get("entry_price", ctx.signal.get("price", 0.0)) or 0.0),
                    sl=float(ctx.signal.get("sl_price", 0.0) or 0.0),
                    tp=float(ctx.signal.get("tp_price", 0.0) or 0.0),
                    symbol=str(ctx.signal.get("symbol", ctx.symbol) or ctx.symbol),
                    source="pipeline",
                )
        elif action_upper in {"CLOSE", "REDUCE"}:
            sync_close_fill(
                ctx.signal_attribution,
                data_dir=ctx.project_root / "data",
                ticket=int(ctx.signal.get("ticket", report.get("deal_id", report.get("order_id", 0))) or 0),
                exit_price=float(report.get("deal_price", ctx.signal.get("entry_price", 0.0)) or 0.0),
                pnl=float(report.get("pnl", 0.0) or 0.0),
                side=str(ctx.signal.get("side", "NONE") or "NONE"),
                reason=str(report.get("reason", "") or ""),
                source="pipeline",
            )
    report["execution_ts"] = str(report.get("execution_ts") or report.get("ts") or datetime.now(timezone.utc).isoformat())
    apply_execution_report(ctx.state_path, ctx.signal_id, report)
    _append_instance_execution_event(ctx.project_root, ctx.signal, report)
    _sync_multi_strategy_blackboard(
        ctx.project_root,
        ctx.scheduler_snapshot,
        ctx.strategy_signals,
        ctx.aggregated_intent,
        report,
    )
    return report


def _dispatch_signal(ctx: PipelineRunContext) -> Optional[Dict[str, Any]]:
    if ctx.state_path is None:
        raise RuntimeError("pipeline state_path is not initialized")
    _apply_live_volume_plan(ctx.signal, project_root=ctx.project_root, aggregated_intent=ctx.aggregated_intent)
    ctx.signal_attribution = _attach_signal_attribution(ctx.signal, ctx.aggregated_intent)
    materialized_action = str(ctx.signal.get("materialized_action", "") or "").upper()
    if materialized_action in {"OPEN_LIMIT", "ARM_WAIT_CONFIRMATION"}:
        transition_states(
            ctx.state_path,
            ctx.signal_id,
            [{"state": "VALIDATED", "metadata": ctx.signal_attribution}],
        )
        report = _register_pending_entry(ctx)
        return _apply_execution_side_effects(ctx, report)
    _sync_request_side_effects(ctx)
    from mt5_execution import generate_intent_id

    intent_id = generate_intent_id(ctx.signal)
    ctx.signal["intent_id"] = intent_id
    transition_states(
        ctx.state_path,
        ctx.signal_id,
        [{"state": "VALIDATED", "metadata": ctx.signal_attribution}],
    )
    report = dispatch_execution(
        ctx.signal,
        ctx.execution_mode,
        ctx.symbol,
        ctx.execution_report_path,
        intent_id=intent_id,
    )
    if report is not None:
        transition_states(
            ctx.state_path,
            ctx.signal_id,
            [{"state": "SENT", "metadata": ctx.signal_attribution}],
        )
    return _apply_execution_side_effects(ctx, report)


def _build_completion_outcome(
    ctx: PipelineRunContext,
    report: Optional[Dict[str, Any]],
) -> PipelineOutcome:
    static_root = ctx.code_root or ctx.project_root
    runbook_path = static_root / "runbooks/incident_playbooks.json"
    metrics = _build_runtime_metrics(ctx.engine_request, ctx.signal)
    incidents = evaluate_incidents(runbook_path, metrics)
    auto_actions = []
    if ctx.enable_incident_autofix:
        for incident in incidents:
            auto_actions.extend(resolve_auto_actions(runbook_path, incident["detector_id"]))
    ledger_events = ["signal_envelope"]
    if ctx.promotion_info.get("observe_only"):
        ledger_events.insert(0, "shadow_observe_only_execution")
    return PipelineOutcome(
        execution_report=report or {},
        incidents=incidents,
        auto_actions=auto_actions,
        ledger_events=ledger_events,
        append_runtime_log=True,
        sync_mailbox=True,
    )


# ═══════════════════════════════════════════════════════════════════════
# run_once — 唯一交易主入口 (Sole Trading Entry Point)
#
# 暂态架构 (2026-05-14):
#   pipeline.py 拥有所有交易裁决权：信号生成与聚合、宏观方向绑定与风控
#   veto、执行状态机推进、是否允许开/平仓、统一日志/ledger/runtime snapshot
#   输出。gold_trader.py 已削减为 --monitor-only 桥接层。
#
# 原则: 单入口、单决策者、单状态机、单事实源、单回报归集器、单写者
# ═══════════════════════════════════════════════════════════════════════
def run_once(
    project_root: Path,
    code_root: Optional[Path],
    tenant_id: str,
    source: str,
    runtime_intent: str,
    input_path: Path,
    output_path: Path,
    execution_mode: str,
    symbol: str,
    execution_report_path: Optional[Path],
    enable_incident_autofix: bool,
    live_client: Optional[Any],
) -> Dict[str, Any]:
    _validate_runtime_intent(source, execution_mode, runtime_intent)
    runtime_state, applied_auto_actions, auto_action_alerts, unsupported_auto_actions = _consume_pending_auto_actions(project_root, tenant_id)
    ctx = PipelineRunContext(
        project_root=project_root,
        code_root=code_root,
        tenant_id=tenant_id,
        source=source,
        runtime_intent=runtime_intent,
        output_path=output_path,
        execution_mode=execution_mode,
        symbol=symbol,
        execution_report_path=execution_report_path,
        enable_incident_autofix=enable_incident_autofix,
        runtime_state=runtime_state,
    )
    if applied_auto_actions:
        ctx.promotion_info["applied_auto_actions"] = applied_auto_actions
    if unsupported_auto_actions:
        ctx.promotion_info["unsupported_auto_actions"] = unsupported_auto_actions
    for index, alert_text in enumerate(auto_action_alerts):
        _mailbox_emit(
            project_root,
            "Risk_Commissar",
            "WARN",
            alert_text,
            f"auto_action_alert:{index}:{alert_text}",
            30.0,
        )
    outcome = _evaluate_runtime_prechecks(ctx, runtime_state=runtime_state)
    if outcome is not None:
        return _emit_pipeline_outcome(ctx, outcome)

    _load_run_context(ctx, input_path=input_path, live_client=live_client)

    outcome = _evaluate_market_data_delay_gate(ctx)
    if outcome is not None:
        return _emit_pipeline_outcome(ctx, outcome)

    _apply_pending_entry_promotions(ctx)

    outcome = _evaluate_retry_outcome(ctx)
    if outcome is not None:
        return _emit_pipeline_outcome(ctx, outcome)

    outcome = _evaluate_guard_outcome(ctx)
    if outcome is not None:
        return _emit_pipeline_outcome(ctx, outcome)

    report = _dispatch_signal(ctx)
    # 不再调用 _sync_agent_state_from_mt5 — agent_state.json 由 gold_trader 唯一拥有。
    # pipeline 只写 execution_state.json（信号决策）和 agent_decision_contract.json（宏观方向）。
    outcome = _build_completion_outcome(ctx, report)
    return _emit_pipeline_outcome(ctx, outcome)


def _cleanup_stale_logs(project_root: Path) -> dict[str, int]:
    """Truncate accumulating log/ledger files to prevent unbounded disk growth.

    Returns counts of truncated/pruned files for observability.
    """
    logs_dir = project_root / "logs"
    if not logs_dir.exists():
        return {}

    stats: dict[str, int] = {}
    now = time.time()
    DAY_SEC = 86400

    # ── ndjson ledgers: keep last N lines ──────────────────────
    ndjson_limits: dict[str, int] = {
        "decision_ledger.ndjson": 10_000,
        "trading_ledger.ndjson": 10_000,
        "restart_fingerprint.ndjson": 2_000,
        "signal_ledger.ndjson": 5_000,
        "risk_events.ndjson": 5_000,
        "decision_signal_links.ndjson": 5_000,
        "shadow_learner_decisions.ndjson": 5_000,
        "live_learner_selections.ndjson": 5_000,
        "autonomy_loop_history.ndjson": 2_000,
    }
    for filename, max_lines in ndjson_limits.items():
        path = logs_dir / filename
        if not path.exists():
            continue
        try:
            lines = path.read_text(encoding="utf-8").splitlines()
            if len(lines) > max_lines:
                path.write_text("\n".join(lines[-max_lines:]) + "\n", encoding="utf-8")
                stats[filename] = len(lines) - max_lines
        except Exception:
            pass

    # ── runtime.log: rotate when > 8K lines, keep last 3 rotations ──
    runtime_log = logs_dir / "runtime.log"
    if runtime_log.exists():
        try:
            lines = runtime_log.read_text(encoding="utf-8").splitlines()
            if len(lines) > 8000:
                ts_tag = time.strftime("%Y%m%d_%H%M%S", time.gmtime())
                archive_path = logs_dir / f"runtime.{ts_tag}.log"
                runtime_log.rename(archive_path)
                stats["runtime.log"] = len(lines)
                # prune old archives beyond 3
                runtime_archives = sorted(logs_dir.glob("runtime.*.log"), key=lambda p: p.stat().st_mtime)
                for old in runtime_archives[:-3]:
                    old.unlink(missing_ok=True)
        except Exception:
            pass

    # ── swarm/*.jsonl: keep last 5K lines each ──────────────────
    swarm_dir = logs_dir / "swarm"
    if swarm_dir.exists():
        for jsonl_file in swarm_dir.glob("*.jsonl"):
            try:
                lines = jsonl_file.read_text(encoding="utf-8").splitlines()
                if len(lines) > 5_000:
                    jsonl_file.write_text("\n".join(lines[-5_000:]) + "\n", encoding="utf-8")
                    stats[f"swarm/{jsonl_file.name}"] = len(lines) - 5_000
            except Exception:
                pass

    # ── gold_trader.*.log: remove files older than 7 days ──────
    for old_log in logs_dir.glob("gold_trader.*.log"):
        try:
            if now - old_log.stat().st_mtime > 7 * DAY_SEC:
                old_log.unlink(missing_ok=True)
                stats[f"pruned:{old_log.name}"] = 1
        except Exception:
            pass

    # ── decision_ledger_archive_*.ndjson: keep last 3 days ─────
    archive_files = sorted(logs_dir.glob("decision_ledger_archive_*.ndjson"),
                           key=lambda p: p.stat().st_mtime)
    for old in archive_files[:-3]:
        try:
            old.unlink(missing_ok=True)
            stats[f"pruned:{old.name}"] = 1
        except Exception:
            pass

    return stats


_CLEANUP_TICK = 0
_CLEANUP_INTERVAL = 300  # every ~300 pipeline cycles (~10 min at 2s poll)

# ── Exit monitoring state (single-cycle, driven by main polling loop) ──
_EXIT_MONITOR_CYCLE = 0
_EXIT_MT5_ABSENT_COUNT = 0


# ═══════════════════════════════════════════════════════════════════════
# Inline exit helpers — small, pure, no external deps needed
# ═══════════════════════════════════════════════════════════════════════

def _check_sltp_conditions(bid: float, ask: float, state: dict) -> tuple:
    """Check if price hit SL/TP. Returns (triggered, reason, action)."""
    if not state.get("position_open"):
        return False, None, None
    side = state.get("position_side", "BUY")
    sl = state.get("position_sl", 0)
    tp = state.get("position_tp", 0)
    spread = round((ask - bid) * 100, 1)
    if spread > 150 or sl <= 0:
        return False, None, None
    if side == "BUY":
        if bid <= sl:
            return True, f"SL bid={bid}<={sl}", "CLOSE"
        if tp > 0 and bid >= tp:
            return True, f"TP bid={bid}>={tp}", "CLOSE"
    else:
        if ask >= sl:
            return True, f"SL ask={ask}>={sl}", "CLOSE"
        if tp > 0 and ask <= tp:
            return True, f"TP ask={ask}<={tp}", "CLOSE"
    return False, None, None


def _seal_close_guard(state: dict, *, closed_side: str = "", closed_ticket: int = 0,
                      close_action: str = "CLOSE", close_ts: float = None) -> None:
    """Mark post-close guard fields to prevent same-side re-entry."""
    if not isinstance(state, dict):
        return
    ts = float(close_ts or time.time())
    state["last_close_time"] = ts
    state["last_closed_ts"] = ts
    if closed_side:
        state["last_closed_side"] = closed_side.upper()
    if closed_ticket:
        state["last_closed_ticket"] = int(closed_ticket)
    state["last_close_action"] = str(close_action or "CLOSE").upper()


def _reset_position_state(state: dict) -> None:
    """Zero all position fields after close (mutates in-place)."""
    for k in ("position_open", "position_ticket", "position_price", "position_sl",
              "position_tp", "position_vol", "position_entry", "position_time"):
        state[k] = 0 if k != "position_open" else False
    state["position_open"] = False
    state["position_side"] = None
    state["position_signal_id"] = ""
    state["position_decision_id"] = ""
    state["position_regime"] = "UNKNOWN"
    state["position_attribution"] = {}
    state["l1_breakeven_done"] = False
    state["partial_done"] = False
    state["trend_run_mode"] = False
    state["trailing_telemetry"] = {}


# MT5 read error counter: after N consecutive failures while position is tracked
# as open, force-close to prevent permanent ghost orders when MT5 files are
# consistently unreadable (EA crash, file lock contention, etc.).
_MT5_SYNC_ERROR_MAX = 5  # ~10 seconds at 2s poll

def _sync_mt5_to_state(state: dict) -> dict:
    """Pull MT5 real position fields into state dict (mutates in-place).

    When MT5 reads fail consistently while state believes a position is open,
    force-close after _MT5_SYNC_ERROR_MAX consecutive failures rather than
    silently preserving stale position state indefinitely (ghost order).
    """
    from bridge.gold_trader import get_mt5_position_state
    mt5 = get_mt5_position_state()
    if mt5.get("error"):
        was_open = bool(state.get("position_open", False))
        if was_open:
            err_count = int(state.get("_mt5_sync_err_count", 0) or 0) + 1
            state["_mt5_sync_err_count"] = err_count
            if err_count >= _MT5_SYNC_ERROR_MAX:
                state["position_open"] = False
                state["_mt5_sync_err_count"] = 0
                state["_mt5_sync_forced_close"] = True
                state["_mt5_sync_forced_close_reason"] = str(mt5.get("error", "unknown"))
        return state
    # MT5 read succeeded — reset error counter
    state["_mt5_sync_err_count"] = 0
    state.pop("_mt5_sync_forced_close", None)
    state.pop("_mt5_sync_forced_close_reason", None)
    for k in ("balance", "equity"):
        if k in mt5 and mt5[k] is not None:
            state[k] = float(mt5[k])
    if bool(mt5.get("position_open")) and int(mt5.get("position_ticket", 0) or 0) > 0:
        for k in ("position_open", "position_ticket", "position_side",
                  "position_price", "position_sl", "position_tp",
                  "position_vol", "position_entry", "positions_all"):
            if k in mt5:
                state[k] = mt5[k]
    else:
        state["position_open"] = False
        if "positions_all" in mt5:
            state["positions_all"] = mt5["positions_all"]
    return state


def _persist_trail_telemetry(telemetry: dict) -> None:
    """Write trailing telemetry snapshot to logs dir."""
    import json as _json
    from bridge.bridge_constants import LOGS_DIR
    dest = LOGS_DIR / "trailing_telemetry_latest.json"
    tmp = dest.with_suffix(".tmp")
    try:
        tmp.write_text(_json.dumps(telemetry, ensure_ascii=False, default=str), encoding="utf-8")
        tmp.replace(dest)
    except Exception:
        pass


def _compute_close_pnl(state_before_close: dict, exit_price: float) -> float:
    """Estimate close PnL from position state."""
    side = str(state_before_close.get("position_side", "BUY") or "BUY").upper()
    entry = float(state_before_close.get("position_entry", 0.0) or 0.0)
    vol = float(state_before_close.get("position_vol", 0.01) or 0.01)
    if entry <= 0 or exit_price <= 0:
        return 0.0
    if side == "BUY":
        return (exit_price - entry) * vol * 100.0
    return (entry - exit_price) * vol * 100.0


def _build_exit_indicators(state: dict, bid: float, ask: float, project_root: Path) -> Dict[str, Any]:
    """Build technical indicators for exit_optimizer.evaluate_exit()."""
    indicators: Dict[str, Any] = {"atr": 3.0}
    try:
        from bridge.strategy import _read_bars
        bars = _read_bars(120)
        if bars and len(bars) >= 26:
            from signal_generators.ema_macd_signal import _adx, _atr, _ema, _macd, _rsi
            closes = [float(b.get("close", 0)) for b in bars[-60:]]
            highs = [float(b.get("high", 0)) for b in bars[-60:]]
            lows = [float(b.get("low", 0)) for b in bars[-60:]]
            indicators["ema9"] = _ema(closes, 9)[-1] if closes else 0
            indicators["ema21"] = _ema(closes, 21)[-1] if closes else 0
            indicators["adx"] = _adx(highs, lows, closes, 14)[-1] if closes else 0
            indicators["rsi"] = _rsi(closes, 14)[-1] if closes else 50
            macd_line, signal_line, hist = _macd(closes)
            indicators["macd"] = macd_line[-1] if macd_line else 0
            indicators["macd_signal"] = signal_line[-1] if signal_line else 0
            indicators["macd_hist"] = hist[-1] if hist else 0
            indicators["atr"] = _atr(highs, lows, closes, 14)[-1] if closes else 3.0
            indicators["bars"] = bars
    except Exception:
        pass
    return indicators


# ═══════════════════════════════════════════════════════════════════════
# Exit monitoring cycle (pipeline owns this, calls gold_trader / exit_optimizer directly)
# ═══════════════════════════════════════════════════════════════════════

def _run_exit_monitoring_cycle(
    state: Dict[str, Any],
    project_root: Path,
    args: argparse.Namespace,
    symbol: str,
    source: str,
    live_client: Any,
    poll_seconds: float,
) -> None:
    """Single-cycle exit monitoring. Pipeline main() calls this every poll_seconds
    while position_open=True.  NOT a while-loop — one iteration per call.

    Calls gold_trader directly for state I/O and EA bridge; calls exit_optimizer
    directly for strategy decisions.
    """
    global _EXIT_MONITOR_CYCLE, _EXIT_MT5_ABSENT_COUNT

    # Direct imports from the modules that own the logic — no wrappers.
    from bridge.gold_trader import (
        _finalize_position_closure,
        _apply_realized_closure_stats,
        _position_owned_by_pipeline,
        get_price,
        load_state,
        process_ea_reports,
        reconcile_positions,
        save_state,
        write_trade_request,
    )
    from bridge.exit_optimizer import evaluate_exit, execute_exit_actions

    _EXIT_MONITOR_CYCLE += 1
    # was_open must account for three cases:
    # 1. Normal: state.position_open is True (MT5 confirms position)
    # 2. Transition: _prev_was_open flag set by polling loop on open→closed transition
    # 3. Mid-count: _mt5_absent_count > 0 means we're confirming a close across cycles
    was_open = (
        bool(state.get("position_open", False))
        or bool(state.get("_prev_was_open", False))
        or (int(state.get("_mt5_absent_count", 0) or 0) > 0)
    )
    state.pop("_prev_was_open", None)
    prev_ticket = int(state.get("position_ticket", 0) or 0)
    prev_side = str(state.get("position_side", "") or "").upper()

    # 1. Process EA reports
    state_before_ea = dict(state)
    pipeline_owned = _position_owned_by_pipeline(state)
    state = process_ea_reports(state)
    if not state.get("position_open") and was_open:
        if pipeline_owned:
            # process_ea_reports skips close bookkeeping for pipeline-owned
            # positions (line 4736); pipeline must fill the gap.
            bid, ask, _, _, _ = get_price()
            exit_price = (bid + ask) / 2.0 if bid and ask else float(
                state_before_ea.get("position_entry", state_before_ea.get("position_price", 0)) or 0
            )
            pos_time = float(state_before_ea.get("position_time", time.time()))
            duration_min = int((time.time() - pos_time) / 60) if pos_time > 0 else 0
            close_pnl = _compute_close_pnl(state_before_ea, exit_price)
            _finalize_position_closure(
                state_before_ea, exit_price, close_pnl, duration_min,
                done_reason="EA_REPORT_CLOSE_DETECTED", close_ts=time.time(),
            )
        _seal_close_guard(state, closed_side=prev_side, closed_ticket=prev_ticket,
                          close_action="EA_REPORT_CLOSE_DETECTED")
        _reset_position_state(state)
        save_state(state)
        _EXIT_MONITOR_CYCLE = 0
        return

    # 2. Sync MT5 position fields
    state = _sync_mt5_to_state(state)
    if state.get("position_open"):
        _EXIT_MT5_ABSENT_COUNT = 0  # MT5 confirms position, reset blink counter
        state["_mt5_absent_count"] = 0  # reset persisted counter too

    # 2a. Forced close: _sync_mt5_to_state hit consecutive MT5 read errors while
    #     tracking an open position — MT5 files are consistently unreadable.
    #     Bypass the 3-count confirmation and close immediately to prevent
    #     permanent ghost orders when EA crashes or files are corrupted.
    if was_open and not state.get("position_open") and state.get("_mt5_sync_forced_close"):
        _EXIT_MT5_ABSENT_COUNT = 0
        state["_mt5_absent_count"] = 0
        bid, ask, _, _, _ = get_price()
        exit_price = (bid + ask) / 2.0 if bid and ask else 0.0
        state_before_close = dict(state)
        close_pnl = _compute_close_pnl(state_before_close, exit_price)
        pos_time = float(state_before_close.get("position_time", time.time()))
        duration_min = int((time.time() - pos_time) / 60) if pos_time > 0 else 0
        forced_reason = str(state.get("_mt5_sync_forced_close_reason", "unknown"))
        _finalize_position_closure(
            state_before_close, exit_price, close_pnl, duration_min,
            done_reason=f"MT5_SYNC_FORCED_CLOSE:{forced_reason}",
            close_ts=time.time(),
        )
        _seal_close_guard(state, closed_side=prev_side, closed_ticket=prev_ticket,
                          close_action="MT5_SYNC_FORCED_CLOSE")
        _reset_position_state(state)
        save_state(state)
        _EXIT_MONITOR_CYCLE = 0
        return

    # 3. Detect external close (MT5 position disappeared).
    #     Use state-persisted _mt5_absent_count so the 3-count confirmation
    #     survives across save_state/load_state cycles and polling-loop
    #     iterations.  The counter is reset to 0 whenever MT5 confirms the
    #     position is still open (step 2 above) or cleanup completes.
    if was_open and not state.get("position_open"):
        absent_count = int(state.get("_mt5_absent_count", 0) or 0) + 1
        state["_mt5_absent_count"] = absent_count
        if absent_count < 3:
            save_state(state)
            return
        state["_mt5_absent_count"] = 0
        bid, ask, _, _, _ = get_price()
        exit_price = (bid + ask) / 2.0 if bid and ask else 0.0
        state_before_close = dict(state)
        close_pnl = _compute_close_pnl(state_before_close, exit_price)
        pos_time = float(state_before_close.get("position_time", time.time()))
        duration_min = int((time.time() - pos_time) / 60) if pos_time > 0 else 0
        _finalize_position_closure(
            state_before_close, exit_price, close_pnl, duration_min,
            done_reason="EXTERNAL_CLOSE_DETECTED",
            close_ts=time.time(),
        )
        _seal_close_guard(state, closed_side=prev_side, closed_ticket=prev_ticket,
                          close_action="EXTERNAL_CLOSE_DETECTED")
        _reset_position_state(state)
        save_state(state)
        return

    if not state.get("position_open"):
        state["_mt5_absent_count"] = 0
        if was_open:
            _seal_close_guard(state, closed_side=prev_side, closed_ticket=prev_ticket,
                              close_action="MONITOR_EXIT_DETECTED")
        save_state(state)
        _EXIT_MONITOR_CYCLE = 0
        return

    # 4. Get current prices
    bid, ask, _, price_source, _ = get_price()
    if bid is None:
        return

    # 5. SL/TP hit check
    triggered, reason, action = _check_sltp_conditions(bid, ask, state)
    if triggered:
        state_before_close = dict(state)
        exit_price = (bid + ask) / 2.0 if bid and ask else 0.0
        pos_time = float(state.get("position_time", time.time()))
        duration_min = int((time.time() - pos_time) / 60) if pos_time > 0 else 0
        write_trade_request(
            action or "CLOSE", state.get("position_side", "BUY"),
            0, 0, 0, state.get("position_vol", 0.01),
            sig_id=str(state.get("position_signal_id") or state.get("position_decision_id") or ""),
            decision_id=str(state.get("position_decision_id") or ""),
            ticket=int(state.get("position_ticket", 0) or 0),
            request_origin="pipeline_sltp",
        )
        time.sleep(5)
        state = load_state()
        state = process_ea_reports(state)
        if not state.get("position_open"):
            if pipeline_owned:
                close_pnl = _compute_close_pnl(state_before_close, exit_price)
                _finalize_position_closure(
                    state_before_close, exit_price, close_pnl, duration_min,
                    done_reason="SLTP_CLOSE_DETECTED", close_ts=time.time(),
                )
            _seal_close_guard(state, closed_side=prev_side, closed_ticket=prev_ticket,
                              close_action="SLTP_CLOSE_DETECTED")
            _reset_position_state(state)
            save_state(state)
            _EXIT_MONITOR_CYCLE = 0
            return

    # 6. Macro escape hatch
    try:
        from bridge.macro_calendar import get_macro_event_signal
        macro_signal = get_macro_event_signal()
        if macro_signal.get("tier") == "BLOCK":
            state_before_close = dict(state)
            exit_price = (bid + ask) / 2.0 if bid and ask else 0.0
            pos_time = float(state.get("position_time", time.time()))
            duration_min = int((time.time() - pos_time) / 60) if pos_time > 0 else 0
            write_trade_request(
                "CLOSE", state.get("position_side", "BUY"),
                0, 0, 0, state.get("position_vol", 0.01),
                sig_id=str(state.get("position_signal_id") or state.get("position_decision_id") or ""),
                decision_id=str(state.get("position_decision_id") or ""),
                ticket=int(state.get("position_ticket", 0) or 0),
                request_origin="pipeline_macro_escape",
            )
            time.sleep(5)
            state = load_state()
            state = process_ea_reports(state)
            if not state.get("position_open"):
                if pipeline_owned:
                    close_pnl = _compute_close_pnl(state_before_close, exit_price)
                    _finalize_position_closure(
                        state_before_close, exit_price, close_pnl, duration_min,
                        done_reason="MACRO_ESCAPE_CLOSE_DETECTED", close_ts=time.time(),
                    )
                _seal_close_guard(state, closed_side=prev_side, closed_ticket=prev_ticket,
                                  close_action="MACRO_ESCAPE_CLOSE_DETECTED")
                _reset_position_state(state)
                save_state(state)
                _EXIT_MONITOR_CYCLE = 0
                return
    except Exception:
        pass

    # 7. Exit optimizer (core strategy)
    try:
        indicators = _build_exit_indicators(state, bid, ask, project_root)
        atr_v = float(indicators.get("atr", 3.0) or 3.0)
        should_exit, reason, exit_detail = evaluate_exit(bid, ask, state, indicators, atr_v)

        if isinstance(exit_detail, dict):
            trail = exit_detail.get("trail_debug", {})
            if isinstance(trail, dict):
                state["trailing_telemetry"] = {**trail, "bid": bid, "ask": ask,
                    "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
                _persist_trail_telemetry(state["trailing_telemetry"])

        if should_exit and isinstance(exit_detail, dict) and "actions" in exit_detail:
            from bridge.bridge_constants import BRIDGE_TO_EA
            state_before_close = dict(state)
            exit_price = (bid + ask) / 2.0 if bid and ask else 0.0
            pos_time = float(state.get("position_time", time.time()))
            duration_min = int((time.time() - pos_time) / 60) if pos_time > 0 else 0
            execute_exit_actions(exit_detail["actions"], state, bid, ask, bridge_to_ea=BRIDGE_TO_EA)
            time.sleep(3)
            state = load_state()
            state = process_ea_reports(state)
            if not state.get("position_open"):
                if pipeline_owned:
                    close_pnl = _compute_close_pnl(state_before_close, exit_price)
                    _finalize_position_closure(
                        state_before_close, exit_price, close_pnl, duration_min,
                        done_reason="EXIT_OPTIMIZER_CLOSE_DETECTED", close_ts=time.time(),
                    )
                _seal_close_guard(state, closed_side=prev_side, closed_ticket=prev_ticket,
                                  close_action="EXIT_OPTIMIZER_CLOSE_DETECTED")
                _reset_position_state(state)
                save_state(state)
                _EXIT_MONITOR_CYCLE = 0
                return
    except Exception:
        import traceback
        try:
            from bridge.gold_trader import _daemon_log
            _daemon_log(f"Exit optimizer cycle error: {traceback.format_exc()}")
        except Exception:
            pass

    # 8. Periodic reconcile (every ~30 cycles ≈ 60s at 2s poll)
    if _EXIT_MONITOR_CYCLE % 30 == 0:
        state = reconcile_positions(state)

    # 9. Persist state
    save_state(state)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input-path", required=True)
    parser.add_argument("--output-path", required=True)
    parser.add_argument("--source", default="file", choices=["file", "mt5", "ea_feed"])
    parser.add_argument("--runtime-intent", default="offline_train", choices=["offline_train", "paper_trade", "live_trade"])
    parser.add_argument("--tenant-id", default="")
    parser.add_argument("--symbol", default="XAUUSD")
    parser.add_argument("--timeframe", default="M5")
    parser.add_argument("--execution-mode", default="file", choices=["file", "mock", "direct", "ea_bridge"])
    parser.add_argument("--execution-report-path", default="")
    parser.add_argument("--poll-seconds", type=float, default=2.0)
    parser.add_argument("--enable-incident-autofix", default="true")
    parser.add_argument("--once", action="store_true")
    args = parser.parse_args()

    # ── License gate ─────────────────────────────────────────────
    try:
        from src.license.guard import enforce_license
        enforce_license()
    except Exception as e:
        print(f"License check failed: {e}", file=sys.stderr)
        return 1

    code_root = Path(__file__).resolve().parents[1]
    project_root = _resolve_runtime_root(code_root)
    input_path = _resolve_runtime_path(project_root, args.input_path)
    output_path = _resolve_runtime_path(project_root, args.output_path)
    execution_report_path = (
        _resolve_runtime_path(project_root, args.execution_report_path)
        if args.execution_report_path
        else None
    )
    enable_incident_autofix = _to_bool(args.enable_incident_autofix)
    live_client = None
    if args.source == "mt5":
        live_client = MT5LiveSource(args.symbol, args.timeframe)
        if not live_client.connect():
            raise SystemExit("MT5 connect failed: source=mt5 requires realtime connection")
    if args.source == "ea_feed":
        live_client = EABridgeLiveSource(args.symbol, args.timeframe, project_root=project_root)
        if not live_client.connect():
            raise SystemExit("EA feed connect failed: source=ea_feed requires bridge market_snapshot.json")

    _acquire_pipeline_singleton_lock(project_root, args.runtime_intent, args.source, args.execution_mode)

    # One-shot log cleanup at startup (defensive: stale files from before this fix).
    _cleanup_stale_logs(project_root)

    if args.once:
        try:
            run_once(
                project_root,
                code_root,
                args.tenant_id,
                args.source,
                args.runtime_intent,
                input_path,
                output_path,
                args.execution_mode,
                args.symbol,
                execution_report_path,
                enable_incident_autofix,
                live_client,
            )
            print(json.dumps({"ok": True, "mode": "once", "output": str(output_path)}, ensure_ascii=False))
            return 0
        finally:
            if live_client is not None:
                live_client.close()
            _release_pipeline_singleton_lock()

    last_mtime = 0.0
    poll_seconds = float(args.poll_seconds)
    global _CLEANUP_TICK
    # Track previous position state across poll iterations so we can detect
    # open→closed transitions.  load_state() syncs with MT5 and may clear
    # position_open before _run_exit_monitoring_cycle gets a chance to run
    # close bookkeeping (finalize/seal/reset).  When a transition is detected,
    # we force one last exit-monitoring cycle with _prev_was_open=True so the
    # 3-count confirmation and cleanup logic can fire.
    _prev_position_open = False
    try:
        while True:
            try:
                # Phase 2: exit monitoring branch — pipeline owns exit orchestration.
                # When position is open, drive exit monitoring directly (no middleman).
                if args.source in {"mt5", "ea_feed"} and args.runtime_intent == "live_trade":
                    from bridge.gold_trader import load_state
                    state = load_state()
                    # Enter exit monitoring when:
                    #  - position is open (normal case), OR
                    #  - we're mid-count confirming an external close (_mt5_absent_count > 0)
                    if state.get("position_open") or state.get("_mt5_absent_count", 0) > 0:
                        # If position just closed (load_state cleared it but we were
                        # tracking one), communicate the transition via _prev_was_open.
                        if not state.get("position_open") and _prev_position_open:
                            state["_prev_was_open"] = True
                        _run_exit_monitoring_cycle(
                            state, project_root, args,
                            args.symbol, args.source, live_client,
                            poll_seconds,
                        )
                        _CLEANUP_TICK += 1
                        if _CLEANUP_TICK >= _CLEANUP_INTERVAL:
                            _cleanup_stale_logs(project_root)
                            _CLEANUP_TICK = 0
                        _prev_position_open = bool(state.get("position_open", False))
                        continue
                    _prev_position_open = False

                if args.source in {"mt5", "ea_feed"}:
                    run_once(
                        project_root,
                        code_root,
                        args.tenant_id,
                        args.source,
                        args.runtime_intent,
                        input_path,
                        output_path,
                        args.execution_mode,
                        args.symbol,
                        execution_report_path,
                        enable_incident_autofix,
                        live_client,
                    )
                    _CLEANUP_TICK += 1
                    if _CLEANUP_TICK >= _CLEANUP_INTERVAL:
                        _cleanup_stale_logs(project_root)
                        _CLEANUP_TICK = 0
                elif input_path.exists():
                    current_mtime = input_path.stat().st_mtime
                    if current_mtime > last_mtime:
                        run_once(
                            project_root,
                            code_root,
                            args.tenant_id,
                            args.source,
                            args.runtime_intent,
                            input_path,
                            output_path,
                            args.execution_mode,
                            args.symbol,
                            execution_report_path,
                            enable_incident_autofix,
                            live_client,
                        )
                        last_mtime = current_mtime
            except RuntimeError as exc:
                # Live feed may be temporarily unavailable during EA/terminal warm-up.
                if args.source in {"mt5", "ea_feed"} and _is_live_source_retryable_error(exc):
                    time.sleep(poll_seconds)
                    continue
                raise
            time.sleep(poll_seconds)
    finally:
        if live_client is not None:
            live_client.close()
        _release_pipeline_singleton_lock()


if __name__ == "__main__":
    raise SystemExit(main())
