from __future__ import annotations

import argparse
import hashlib
import hmac
import json
import os
import sys
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from urllib.parse import parse_qs, urlparse

try:
    from src.chat_tuning import apply_chat_command
    from src.config_snapshot import create_snapshot, list_snapshots, rollback_snapshot
    from src.tenant_manager import get_audit_path, list_tenants, load_runtime_state, tenant_paths
    from src.webhook_adapter import normalize_webhook
    from src.metrics_agg import build_tenant_metrics, evaluate_rollback_gates
    from src.system_bus import update_system_section
except ModuleNotFoundError:
    from chat_tuning import apply_chat_command
    from config_snapshot import create_snapshot, list_snapshots, rollback_snapshot
    from tenant_manager import get_audit_path, list_tenants, load_runtime_state, tenant_paths
    from webhook_adapter import normalize_webhook
    from metrics_agg import build_tenant_metrics, evaluate_rollback_gates
    from system_bus import update_system_section


def _json_response(handler: BaseHTTPRequestHandler, code: int, payload: Dict[str, Any]) -> None:
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    handler.send_response(code)
    handler.send_header("Content-Type", "application/json; charset=utf-8")
    handler.send_header("Content-Length", str(len(body)))
    handler.end_headers()
    handler.wfile.write(body)


def _read_json_body(handler: BaseHTTPRequestHandler) -> Tuple[bytes, Dict[str, Any]]:
    length = int(handler.headers.get("Content-Length", "0"))
    data = handler.rfile.read(length) if length > 0 else b"{}"
    return data, json.loads(data.decode("utf-8"))


class ControlHandler(BaseHTTPRequestHandler):
    project_root: Path
    control_api_token: str
    feishu_webhook_token: str
    qq_webhook_token: str
    feishu_webhook_secret: str
    qq_webhook_secret: str
    rate_limit_per_minute: int
    idempotency_ttl_seconds: int
    webhook_max_skew_seconds: int
    webhook_nonce_ttl_seconds: int
    rate_store: Dict[str, Dict[str, Any]]
    idempotency_store: Dict[str, Dict[str, Any]]
    webhook_nonce_store: Dict[str, Dict[str, Any]]

    def _auth_ok(self) -> bool:
        token = self.headers.get("X-Api-Key", "")
        return bool(self.control_api_token) and token == self.control_api_token

    def _tenant_id(self, body: Optional[Dict[str, Any]] = None) -> str:
        header_tenant = self.headers.get("X-Tenant", "").strip()
        if header_tenant:
            return header_tenant
        if body is not None:
            return str(body.get("tenant_id", "")).strip()
        return ""

    def _extract_text_from_webhook(self, channel: str, body: Dict[str, Any]) -> str:
        if channel == "feishu":
            msg = body.get("event", {}).get("message", {})
            content_raw = msg.get("content", "")
            if isinstance(content_raw, str):
                try:
                    payload = json.loads(content_raw)
                    return str(payload.get("text", "")).strip()
                except Exception:
                    return content_raw.strip()
        if channel == "qq":
            return str(body.get("text", "")).strip()
        return ""

    def _webhook_auth_ok(self, channel: str, query: Dict[str, list[str]], body: Dict[str, Any]) -> bool:
        expected = self.feishu_webhook_token if channel == "feishu" else self.qq_webhook_token
        if not expected:
            return True
        token = self.headers.get("X-Webhook-Token", "").strip()
        if not token:
            token = str(body.get("token", "")).strip()
        if not token:
            values = query.get("token", [])
            token = values[0].strip() if values else ""
        return token == expected

    def _verify_webhook_signature(self, channel: str, raw_body: bytes) -> bool:
        if channel == "feishu":
            secret = self.feishu_webhook_secret
            if not secret:
                return True
            ts = self.headers.get("X-Timestamp", "").strip()
            sig = self.headers.get("X-Signature", "").strip()
            if not ts or not sig:
                return False
            payload = (ts + "\n").encode("utf-8") + raw_body
            expected = hmac.new(secret.encode("utf-8"), payload, hashlib.sha256).hexdigest()
            return hmac.compare_digest(expected, sig)
        secret = self.qq_webhook_secret
        if not secret:
            return True
        sig = self.headers.get("X-Signature", "").strip()
        if not sig:
            return False
        expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
        return hmac.compare_digest(expected, sig)

    def _webhook_ts_valid(self) -> bool:
        ts = self.headers.get("X-Timestamp", "").strip()
        if not ts:
            return True
        try:
            iv = int(ts)
        except Exception:
            return False
        now = int(time.time())
        return abs(now - iv) <= int(self.webhook_max_skew_seconds)

    def _webhook_nonce_seen(self, channel: str) -> bool:
        nonce = self.headers.get("X-Nonce", "").strip()
        if not nonce:
            return False
        self._cleanup_runtime_cache()
        key = f"{channel}:{nonce}"
        return key in self.webhook_nonce_store

    def _record_webhook_nonce(self, channel: str) -> None:
        nonce = self.headers.get("X-Nonce", "").strip()
        if not nonce:
            return
        key = f"{channel}:{nonce}"
        self.webhook_nonce_store[key] = {"ts": time.time()}

    def _cleanup_runtime_cache(self) -> None:
        now = time.time()
        rate_ttl = float(getattr(self, 'rate_limit_per_minute', 1)) * 60
        for key in list(self.rate_store.keys()):
            if now - float(self.rate_store[key].get("ts", 0)) > rate_ttl:
                self.rate_store.pop(key, None)
        for key in list(self.idempotency_store.keys()):
            if now - float(self.idempotency_store[key].get("ts", 0)) > float(self.idempotency_ttl_seconds):
                self.idempotency_store.pop(key, None)
        for key in list(self.webhook_nonce_store.keys()):
            if now - float(self.webhook_nonce_store[key].get("ts", 0)) > float(self.webhook_nonce_ttl_seconds):
                self.webhook_nonce_store.pop(key, None)

    def _allow_rate(self, tenant_id: str, endpoint: str) -> bool:
        self._cleanup_runtime_cache()
        now = time.time()
        minute = int(now // 60)
        key = f"{tenant_id or 'default'}:{endpoint}:{minute}"
        item = self.rate_store.get(key, {"count": 0, "ts": now})
        item["count"] = int(item.get("count", 0)) + 1
        item["ts"] = now
        self.rate_store[key] = item
        return int(item["count"]) <= int(self.rate_limit_per_minute)

    def _idempotency_key(self, body: Dict[str, Any]) -> str:
        header_key = self.headers.get("X-Idempotency-Key", "").strip()
        if header_key:
            return header_key
        return str(body.get("request_id", "")).strip()

    def _idempotency_get(self, key: str) -> Optional[Dict[str, Any]]:
        if not key:
            return None
        self._cleanup_runtime_cache()
        item = self.idempotency_store.get(key)
        if not item:
            return None
        return item.get("response")

    def _idempotency_set(self, key: str, response: Dict[str, Any]) -> None:
        if not key:
            return
        self.idempotency_store[key] = {"ts": time.time(), "response": response}

    def _audit_export(self, tenant_id: str, limit: int, offset: int, category: str, since_ts: int, until_ts: int) -> Dict[str, Any]:
        audit_path = get_audit_path(self.project_root, tenant_id)
        if not audit_path.exists():
            return {"ok": True, "tenant_id": tenant_id or "default", "records": []}
        lines = audit_path.read_text(encoding="utf-8").splitlines()
        lines = lines[offset:] if offset > 0 else lines
        selected = lines[-limit:] if limit > 0 else lines
        records = []
        for line in selected:
            try:
                record = json.loads(line)
                if category and str(record.get("intent", "")) != category:
                    continue
                ts_str = str(record.get("timestamp", ""))
                if ts_str:
                    try:
                        ts_norm = ts_str.replace("Z", "+00:00")
                        ts = int(datetime.fromisoformat(ts_norm).astimezone(timezone.utc).timestamp())
                    except Exception:
                        ts = 0
                    if since_ts > 0 and ts < since_ts:
                        continue
                    if until_ts > 0 and ts > until_ts:
                        continue
                records.append(record)
            except Exception:
                continue
        return {
            "ok": True,
            "tenant_id": tenant_id or "default",
            "records": records,
            "total_lines": len(lines),
            "offset": offset,
            "limit": limit,
        }

    def _tenant_metrics(self, tenant_id: str) -> Dict[str, Any]:
        """Build real tenant metrics from logs and runtime state."""
        default_key = tenant_id or "default"
        rate_keys = [k for k in self.rate_store.keys() if k.startswith(f"{default_key}:")]
        idm_keys = list(self.idempotency_store.keys())
        try:
            agg = build_tenant_metrics(self.project_root, tenant_id)
        except Exception:
            agg = {}
        triggered = evaluate_rollback_gates(agg.get("execution", {}))
        return {
            "tenant_id": default_key,
            "execution": agg.get("execution", {}),
            "decision": {
                "window_15m": agg.get("window_15m", {}),
                "window_1h": agg.get("window_1h", {}),
                "window_1d": agg.get("window_1d", {}),
                "error_rate_pct_15m": agg.get("decision", {}).get("decision_error_rate_pct_15m", 0),
                "approval_rate_pct_1d": agg.get("decision", {}).get("decision_approval_rate_pct_1d", 0),
            },
            "runtime": agg.get("runtime", {}),
            "goalTargets": agg.get("goal_targets", {}),
            "rollback_triggers": triggered,
            "rollbacks_pending": len(triggered),
            "cache": {
                "rate_keys": len(rate_keys),
                "idempotency_cache_size": len(idm_keys),
                "webhook_nonce_cache_size": len(self.webhook_nonce_store),
            },
        }

    def _handle_stripe_webhook(self) -> None:
        try:
            from src.payment.stripe import handle_webhook
            length = int(self.headers.get("Content-Length", "0"))
            payload = self.rfile.read(length) if length > 0 else b"{}"
            signature = self.headers.get("Stripe-Signature", "")
            result = handle_webhook(payload, signature)
            _json_response(self, 200 if result.get("ok") else 400, result)
        except ImportError:
            _json_response(self, 500, {"ok": False, "error": "Payment module not available"})
        except Exception as e:
            _json_response(self, 500, {"ok": False, "error": str(e)})

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        path = parsed.path
        # ── Auth routes (public + authenticated) ──
        try:
            from src.auth.auth_api import handle_auth_routes
            if handle_auth_routes(self, path):
                return
        except ImportError:
            pass
        if path == "/health":
            _json_response(self, 200, {"ok": True, "service": "gold-control-api"})
            return
        if path == "/health/full":
            try:
                from src.health.dashboard import get_full_health
                include_disk = parse_qs(parsed.query).get("disk", ["0"])[0] == "1"
                _json_response(self, 200, get_full_health(include_disk=include_disk))
            except ImportError:
                _json_response(self, 500, {"ok": False, "error": "Health module not available"})
            return
        if path == "/config/api-key" and self.command == "GET":
            try:
                from src.llm.config import load_llm_providers
                providers = load_llm_providers()
                deepseek = providers.get("deepseek", {})
                api_key = deepseek.get("apiKey", "")
                has_key = bool(api_key and api_key != "YOUR_DEEPSEEK_API_KEY_HERE")
                _json_response(self, 200, {
                    "ok": True,
                    "configured": has_key,
                    "masked": (api_key[:8] + "****" + api_key[-4:]) if has_key and len(api_key) >= 12 else "",
                })
            except Exception as e:
                _json_response(self, 500, {"ok": False, "error": str(e)})
            return
        if not self._auth_ok():
            _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
            return
        if path == "/config/api-key" and self.command == "POST":
            try:
                _, body = _read_json_body(self)
                api_key = str(body.get("api_key", "") or "").strip()
                if not api_key:
                    _json_response(self, 400, {"ok": False, "error": "api_key is required"})
                    return
                import config_manager
                config_manager.save_provider_api_key("deepseek", api_key)
                os.environ["DEEPSEEK_API_KEY"] = api_key
                _json_response(self, 200, {"ok": True, "configured": True})
            except Exception as e:
                _json_response(self, 500, {"ok": False, "error": str(e)})
            return
        if path == "/runtime":
            tenant_id = self._tenant_id()
            payload = load_runtime_state(self.project_root, tenant_id)
            _json_response(self, 200, {"ok": True, "tenant_id": tenant_id or "default", "runtime": payload})
            return
        if path == "/runtime/tenants":
            tenants = list_tenants(self.project_root)
            _json_response(self, 200, {"ok": True, "tenants": tenants})
            return
        if path == "/runtime/metrics":
            tenant_id = parse_qs(parsed.query).get("tenant_id", [""])[0]
            _json_response(self, 200, {"ok": True, "metrics": self._tenant_metrics(tenant_id)})
            return
        if path == "/audit/export":
            query = parse_qs(parsed.query)
            tenant_id = query.get("tenant_id", [""])[0].strip()
            try:
                limit = int(query.get("limit", ["100"])[0])
            except Exception:
                limit = 100
            try:
                offset = int(query.get("offset", ["0"])[0])
            except Exception:
                offset = 0
            category = query.get("category", [""])[0].strip()
            try:
                since_ts = int(query.get("since_ts", ["0"])[0])
            except Exception:
                since_ts = 0
            try:
                until_ts = int(query.get("until_ts", ["0"])[0])
            except Exception:
                until_ts = 0
            payload = self._audit_export(tenant_id, max(1, min(limit, 2000)), max(0, offset), category, since_ts, until_ts)
            _json_response(self, 200, payload)
            return
        if path == "/telemetry/errors":
            try:
                from src.telemetry.tracker import get_error_report, get_error_stats
                query = parse_qs(parsed.query)
                limit = int(query.get("limit", ["50"])[0])
                etype = query.get("type", [""])[0]
                _json_response(self, 200, {"ok": True, "errors": get_error_report(limit, etype), "stats": get_error_stats()})
            except ImportError:
                _json_response(self, 500, {"ok": False, "error": "Telemetry module not available"})
            return
        if path == "/positions":
            try:
                import json as _json
                import os as _os2
                import pathlib
                # EA writes to FILE_COMMON location
                if sys.platform == "win32":
                    MT5_FILES = pathlib.Path(_os2.environ["APPDATA"]) / "MetaQuotes" / "Terminal" / "Common" / "Files"
                else:
                    MT5_FILES = pathlib.Path.home() / "Library/Application Support/net.metaquotes.wine.metatrader5/drive_c/users/user/AppData/Roaming/MetaQuotes/Terminal/Common/Files"
                BRIDGE_FROM_EA = MT5_FILES / "gold_autotrade/bridge/from_ea"
                
                positions = []
                
                # Check terminal_status.json for current positions
                status_file = MT5_FILES / "gold_autotrade/bridge/terminal_status.json"
                if status_file.exists():
                    try:
                        data = _json.loads(status_file.read_text(encoding="utf-8"))
                        if data.get("positions"):
                            positions.extend(data.get("positions", []))
                    except:
                        pass
                
                _json_response(self, 200, {"ok": True, "positions": positions, "count": len(positions), "updated": data.get("updated", "unknown") if status_file.exists() else "file not found"})
            except Exception as exc:
                _json_response(self, 500, {"ok": False, "reason": str(exc)})
            return
        # ── Data pipeline query endpoints ────────────────────────────────
        elif path.startswith("/data_pipeline"):
            if not self._auth_ok():
                _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
                return
            sub = path.removeprefix("/data_pipeline")
            if sub == "/status":
                try:
                    from src.data_pipeline.monitor import get_pipeline_status
                    _json_response(self, 200, get_pipeline_status())
                except Exception as exc:
                    _json_response(self, 500, {"ok": False, "reason": str(exc)})
                return
            elif sub == "/upload_status":
                try:
                    from src.data_pipeline.event_capture import get_upload_status
                    _json_response(self, 200, get_upload_status())
                except Exception as exc:
                    _json_response(self, 500, {"ok": False, "reason": str(exc)})
                return
            elif sub == "/quality":
                try:
                    from src.data_pipeline.event_capture import get_new_trade_events
                    events, _ = get_new_trade_events(batch_size=20)
                    from src.data_pipeline.quality import validate_batch
                    result = validate_batch(events[:20])
                    _json_response(self, 200, result)
                except Exception as exc:
                    _json_response(self, 500, {"ok": False, "reason": str(exc)})
                return
            elif sub == "/progress":
                # GET /data_pipeline/progress?batch_id=xxx
                batch_id = query.get("batch_id", [None])[0]
                tenant_id_q = query.get("tenant_id", ["default"])[0]
                try:
                    from src.data_pipeline.async_uploader import get_upload_progress, check_upload_circuit
                    if batch_id:
                        prog = get_upload_progress(tenant_id_q or "default", batch_id)
                        if prog is None:
                            _json_response(self, 404, {"ok": False, "reason": "batch_not_found"})
                        else:
                            _json_response(self, 200, prog)
                    else:
                        circuit = check_upload_circuit()
                        _json_response(self, 200, {
                            "circuit": circuit.get("circuit", {}),
                            "adaptive_batcher": circuit.get("batcher", {}),
                        })
                except Exception as exc:
                    _json_response(self, 500, {"ok": False, "reason": str(exc)})
                return
            elif sub == "/circuit":
                try:
                    from src.data_pipeline.async_uploader import check_upload_circuit
                    _json_response(self, 200, check_upload_circuit())
                except Exception as exc:
                    _json_response(self, 500, {"ok": False, "reason": str(exc)})
                return

        _json_response(self, 404, {"ok": False, "reason": "NOT_FOUND"})

    def do_POST(self) -> None:
        parsed = urlparse(self.path)
        path = parsed.path
        query = parse_qs(parsed.query)
        # ── Auth routes (public + authenticated) ──
        try:
            from src.auth.auth_api import handle_auth_routes
            if handle_auth_routes(self, path):
                return
        except ImportError:
            pass
        # ── Stripe webhook (public, signature-verified) ──
        if path == "/stripe/webhook":
            self._handle_stripe_webhook()
            return
        if path == "/tune":
            if not self._auth_ok():
                _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
                return
            _, body = _read_json_body(self)
            idem = self._idempotency_key(body)
            cached = self._idempotency_get(idem)
            if cached is not None:
                _json_response(self, 200, cached)
                return
            role = str(body.get("role", "viewer"))
            channel = str(body.get("channel", "feishu"))
            text = str(body.get("text", ""))
            approved = bool(body.get("approved", False))
            tenant_id = self._tenant_id(body)
            if not self._allow_rate(tenant_id, "tune"):
                _json_response(self, 429, {"ok": False, "reason": "RATE_LIMITED"})
                return
            result = apply_chat_command(self.project_root, role, channel, text, approved, tenant_id)
            self._idempotency_set(idem, result)
            _json_response(self, 200 if result.get("ok") else 400, result)
            return
        if path == "/webhook/feishu":
            raw_body, body = _read_json_body(self)
            if not self._webhook_auth_ok("feishu", query, body):
                _json_response(self, 401, {"ok": False, "reason": "WEBHOOK_UNAUTHORIZED"})
                return
            if not self._verify_webhook_signature("feishu", raw_body):
                _json_response(self, 401, {"ok": False, "reason": "WEBHOOK_BAD_SIGNATURE"})
                return
            if not self._webhook_ts_valid():
                _json_response(self, 401, {"ok": False, "reason": "WEBHOOK_TIMESTAMP_INVALID"})
                return
            if self._webhook_nonce_seen("feishu"):
                _json_response(self, 409, {"ok": False, "reason": "WEBHOOK_REPLAY_DETECTED"})
                return
            challenge = body.get("challenge")
            if challenge:
                _json_response(self, 200, {"challenge": challenge})
                return
            normalized_ok, normalized = normalize_webhook(
                self.project_root,
                "feishu",
                body,
                {"X-Tenant": self.headers.get("X-Tenant", "")},
            )
            if not normalized_ok:
                _json_response(self, 400, {"ok": False, "reason": normalized.get("reason", "WEBHOOK_NORMALIZE_FAILED")})
                return
            tenant_id = normalized["tenant_id"]
            if not self._allow_rate(tenant_id, "webhook_feishu"):
                _json_response(self, 429, {"ok": False, "reason": "RATE_LIMITED"})
                return
            result = apply_chat_command(
                self.project_root,
                str(normalized["role"]),
                str(normalized["channel"]),
                str(normalized["text"]),
                bool(normalized["approved"]),
                tenant_id,
            )
            self._record_webhook_nonce("feishu")
            _json_response(self, 200 if result.get("ok") else 400, result)
            return
        if path == "/webhook/qq":
            raw_body, body = _read_json_body(self)
            if not self._webhook_auth_ok("qq", query, body):
                _json_response(self, 401, {"ok": False, "reason": "WEBHOOK_UNAUTHORIZED"})
                return
            if not self._verify_webhook_signature("qq", raw_body):
                _json_response(self, 401, {"ok": False, "reason": "WEBHOOK_BAD_SIGNATURE"})
                return
            if not self._webhook_ts_valid():
                _json_response(self, 401, {"ok": False, "reason": "WEBHOOK_TIMESTAMP_INVALID"})
                return
            if self._webhook_nonce_seen("qq"):
                _json_response(self, 409, {"ok": False, "reason": "WEBHOOK_REPLAY_DETECTED"})
                return
            normalized_ok, normalized = normalize_webhook(
                self.project_root,
                "qq",
                body,
                {"X-Tenant": self.headers.get("X-Tenant", "")},
            )
            if not normalized_ok:
                _json_response(self, 400, {"ok": False, "reason": normalized.get("reason", "WEBHOOK_NORMALIZE_FAILED")})
                return
            tenant_id = normalized["tenant_id"]
            if not self._allow_rate(tenant_id, "webhook_qq"):
                _json_response(self, 429, {"ok": False, "reason": "RATE_LIMITED"})
                return
            result = apply_chat_command(
                self.project_root,
                str(normalized["role"]),
                str(normalized["channel"]),
                str(normalized["text"]),
                bool(normalized["approved"]),
                tenant_id,
            )
            self._record_webhook_nonce("qq")
            _json_response(self, 200 if result.get("ok") else 400, result)
            return
        if path == "/config/snapshot/create":
            if not self._auth_ok():
                _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
                return
            _, body = _read_json_body(self)
            tenant_id = self._tenant_id(body)
            reason = str(body.get("reason", "")).strip()
            payload = create_snapshot(self.project_root, tenant_id, reason)
            _json_response(self, 200, {"ok": True, "snapshot": payload})
            return
        if path == "/config/snapshot/list":
            if not self._auth_ok():
                _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
                return
            _, body = _read_json_body(self)
            tenant_id = self._tenant_id(body)
            payload = list_snapshots(self.project_root, tenant_id)
            _json_response(self, 200, {"ok": True, "snapshot": payload})
            return
        if path == "/config/snapshot/rollback":
            if not self._auth_ok():
                _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
                return
            _, body = _read_json_body(self)
            tenant_id = self._tenant_id(body)
            tag = str(body.get("tag", "")).strip()
            if not tag:
                _json_response(self, 400, {"ok": False, "reason": "TAG_REQUIRED"})
                return
            try:
                payload = rollback_snapshot(self.project_root, tenant_id, tag)
            except Exception as exc:
                _json_response(self, 400, {"ok": False, "reason": str(exc)})
                return
            _json_response(self, 200, {"ok": True, "snapshot": payload})
            return
        if path == "/runtime/stop":
            if not self._auth_ok():
                _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
                return
            _, body = _read_json_body(self)
            tenant_id = self._tenant_id(body)
            runtime_path = tenant_paths(self.project_root, tenant_id)["runtime"]
            payload = json.loads(runtime_path.read_text(encoding="utf-8")) if runtime_path.exists() else {}
            payload["kill_switch"] = True
            runtime_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
            try:
                update_system_section(
                    root=self.project_root,
                    tenant_id=tenant_id,
                    section="runtime",
                    patch={"kill_switch": True, "kill_switch_updated_at": datetime.now(timezone.utc).isoformat()},
                    source="control_api",
                    event_type="kill_switch_enabled",
                )
            except Exception:
                pass
            _json_response(self, 200, {"ok": True, "tenant_id": tenant_id or "default", "runtime": payload})
            return
        if path == "/positions/close":
            if not self._auth_ok():
                _json_response(self, 401, {"ok": False, "reason": "UNAUTHORIZED"})
                return
            _, body = _read_json_body(self)
            ticket = body.get("ticket")
            symbol = body.get("symbol")
            try:
                import json as _json
                import pathlib
                import time
                # EA reads from FILE_COMMON location
                import os as _os3
                if sys.platform == "win32":
                    MT5_FILES = pathlib.Path(_os3.environ["APPDATA"]) / "MetaQuotes" / "Terminal" / "Common" / "Files"
                else:
                    MT5_FILES = pathlib.Path.home() / "Library/Application Support/net.metaquotes.wine.metatrader5/drive_c/users/user/AppData/Roaming/MetaQuotes/Terminal/Common/Files"
                BRIDGE_TO_EA = MT5_FILES / "gold_autotrade/bridge/to_ea"
                BRIDGE_TO_EA.mkdir(parents=True, exist_ok=True)
                close_request = {
                    "action": "CLOSE",
                    "ticket": ticket,
                    "symbol": symbol or "XAUUSD",
                    "timestamp": time.time(),
                }
                req_file = BRIDGE_TO_EA / f"close_{int(time.time()*1000)}.request.json"
                req_file.write_text(_json.dumps(close_request))
                _json_response(self, 200, {"ok": True, "message": "Close request sent", "request": close_request})
            except Exception as exc:
                _json_response(self, 500, {"ok": False, "reason": str(exc)})
            return
        _json_response(self, 404, {"ok": False, "reason": "NOT_FOUND"})

    def log_message(self, format: str, *args: Any) -> None:
        return


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8787)
    parser.add_argument("--token", default="")
    parser.add_argument("--rate-limit-per-minute", type=int, default=60)
    parser.add_argument("--idempotency-ttl-seconds", type=int, default=600)
    parser.add_argument("--webhook-max-skew-seconds", type=int, default=300)
    parser.add_argument("--webhook-nonce-ttl-seconds", type=int, default=600)
    args = parser.parse_args()

    project_root = Path(__file__).resolve().parents[1]
    token = args.token or os.environ.get("CONTROL_API_TOKEN", "")
    feishu_webhook_token = os.environ.get("FEISHU_WEBHOOK_TOKEN", "")
    qq_webhook_token = os.environ.get("QQ_WEBHOOK_TOKEN", "")
    feishu_webhook_secret = os.environ.get("FEISHU_WEBHOOK_SECRET", "")
    qq_webhook_secret = os.environ.get("QQ_WEBHOOK_SECRET", "")
    handler_cls = type(
        "RuntimeControlHandler",
        (ControlHandler,),
        {
            "project_root": project_root,
            "control_api_token": token,
            "feishu_webhook_token": feishu_webhook_token,
            "qq_webhook_token": qq_webhook_token,
            "feishu_webhook_secret": feishu_webhook_secret,
            "qq_webhook_secret": qq_webhook_secret,
            "rate_limit_per_minute": int(args.rate_limit_per_minute),
            "idempotency_ttl_seconds": int(args.idempotency_ttl_seconds),
            "webhook_max_skew_seconds": int(args.webhook_max_skew_seconds),
            "webhook_nonce_ttl_seconds": int(args.webhook_nonce_ttl_seconds),
            "rate_store": {},
            "idempotency_store": {},
            "webhook_nonce_store": {},
        },
    )
    server = HTTPServer((args.host, args.port), handler_cls)
    print(json.dumps({"ok": True, "host": args.host, "port": args.port}, ensure_ascii=False))
    server.serve_forever()
    return 0


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