modules/bot_api_server.py

"""
modules/bot_api_server.py — direct HTTP API the AWS bot exposes (v18.0).

Replaces the old push-heartbeat/poll-relay pair (bot.py's
_status_heartbeat_loop/_command_poll_loop/_handle_query_command, all
deleted — see bot.py's v18.0 changelog) that used to bounce every dashboard
interaction through a Cloudflare Worker's /status, /command and
/command-result routes. That Worker is retired; this bot now runs
standalone on AWS (Frankfurt) and serves its own status/control API
directly, which the Hugging Face dashboard (app.py, DASHBOARD_ONLY mode)
calls over the internet using BOT_API_URL + BOT_API_TOKEN instead of
ORACLE_URL + RELAY_AUTH_TOKEN.

Endpoints:
  GET  /status   — live-computed JSON status snapshot (no auth — same
                   shape the old /status push payload had, plus a new
                   jupiter_tier field). Computed fresh on every request,
                   never cached.
  POST /command  — Bearer-token gated (BOT_API_TOKEN). Body:
                   {"action": str, "params": dict|null}. Routes `action`
                   to the same handler functions the old poll loop called,
                   but SYNCHRONOUSLY — one HTTP request/response, no more
                   request_id + second poll for the result. Returns
                   {"ok": true, "result": "..."} or {"ok": false, "error":
                   "..."}.

Run as an asyncio background task from bot.py (self._spawn_background),
started unconditionally — with no BOT_API_TOKEN configured, /command
always 503s (logged once) but /status still works, since it carries no
secrets.
"""

from __future__ import annotations

import asyncio
import hmac
import logging
import os
import time
from typing import Any

logger = logging.getLogger(__name__)

_DEFAULT_HOST = "0.0.0.0"
_DEFAULT_PORT = 8081

# Fire-and-forget actions — reuse the exact same CommandHandlers methods
# Telegram's /ghost_mode, /mint_mode and /hunt already call, and report to
# the same Telegram chat, so a dashboard-triggered action leaves the same
# audit trail a Telegram command would. Each takes (chat_id) and sends its
# own Telegram message; the HTTP response below just confirms it ran.
_FIRE_AND_FORGET_ACTIONS = ("toggle_ghost", "toggle_mint", "hunt")

# Data-query actions — rendered via the shared modules/dashboard_panels.py
# renderers (same ones app.py's non-DASHBOARD_ONLY path and the /speedtest
# Telegram command use) and returned directly in the HTTP response body.
_QUERY_ACTIONS = (
    "wallet_status", "gas_health", "price_check", "scanner_tokens",
    "solana_prices", "solana_routes", "arbitrage_detail", "audit_log",
    "speedtest",
)


def _bot_api_token() -> str:
    return os.getenv("BOT_API_TOKEN", "").strip()


def _note_auth(request: Any, ok: bool) -> None:
    """Report an authentication outcome to the HTTP guard (v18.2).

    Best-effort by design: if the guard is unavailable or raises, the auth
    decision itself has already been made and must stand. This can only
    ever add a future reason to refuse, never grant access.
    """
    try:
        from modules.http_guard import client_key, get_guard

        guard = get_guard()
        key = request.get("client_key") or client_key(request)
        if ok:
            guard.note_auth_success(key)
        else:
            guard.note_auth_failure(key)
    except Exception as exc:  # noqa: BLE001
        logger.debug("[BotAPI] could not record auth outcome: %s", exc)


def _bot_api_host() -> str:
    return os.getenv("BOT_API_HOST", _DEFAULT_HOST).strip() or _DEFAULT_HOST


def _bot_api_port() -> int:
    try:
        return int(os.getenv("BOT_API_PORT", str(_DEFAULT_PORT)).strip())
    except ValueError:
        return _DEFAULT_PORT


def _auto_hunt_enabled() -> bool:
    """Same semantics as bot.py's own _auto_hunt_enabled() — duplicated as
    a tiny env-var read rather than imported, to avoid a circular import
    (bot.py imports this module at startup, before its own module-level
    code has finished running)."""
    return os.environ.get("AUTO_HUNT_ENABLED", "false").strip().lower() in (
        "1", "true", "yes", "on",
    )


async def _build_status_snapshot(bot: Any) -> dict[str, Any]:
    """Same fields the old _status_heartbeat_loop's push payload had, plus
    jupiter_tier (new). Computed fresh every call — no caching."""
    from constants import jupiter_tier_status

    h = getattr(bot, "_handlers", None)
    payout = getattr(bot, "_payout", None)
    ledger = await payout.get_status() if payout is not None else None
    solana = getattr(bot, "_solana", None)

    _is_paid, tier_name = jupiter_tier_status()

    return {
        "uptime_s":           int(time.time() - getattr(bot, "start_time", time.time())),
        "telegram_connected": bool(getattr(getattr(bot, "_tg", None), "reachable", False)),
        "ghost_mode":         bool(getattr(h, "_ghost_mode", False)) if h else False,
        "mint_mode":          bool(getattr(h, "_mint_mode", False)) if h else False,
        "auto_hunt":          _auto_hunt_enabled(),
        "min_profit_arb":     h.effective_min_profit if h else getattr(bot.config, "min_profit_usd", None),
        # FIX — this used to independently re-read SOLANA_MIN_PROFIT_USD
        # with its own hardcoded default (2.0, different from
        # solana_arb.py's own default), the same class of banner-vs-
        # enforced-value drift already fixed once for mint mode (see
        # modules/command_handlers.py's v17.20 FIX). Now reads the LIVE
        # SolanaArbEngine instance's own .min_profit attribute so the
        # displayed value can never drift from what's actually enforced.
        "min_profit_sol":     solana.min_profit if solana is not None else None,
        "total_profit_usd":   (ledger or {}).get("net_profit", 0.0),
        "scans_total":        (ledger or {}).get("scan_count", 0),
        "scans_buy":          (ledger or {}).get("buy_count", 0),
        "dry_run":            getattr(bot.config, "dry_run", None),
        "jupiter_tier":       tier_name,
        # v18.1 — operational fields for the public page. Audited as
        # non-secret before being added here, because this endpoint is
        # unauthenticated and fronted by bot.elghaly.dev: scan age is a
        # timestamp delta, route counts are integers, and the governor's
        # rate is a number about our own pacing. No URL, no address, no
        # key, no config value that would help an attacker. The RPC URL in
        # particular stays OUT — SOLANA_RPC_PRIMARY embeds a provider API
        # key in its path on this deployment.
        "solana": ({
            "last_scan_age_secs": (
                round(time.time() - solana.last_scan_ts, 1) if solana.last_scan_ts else None
            ),
            "scan_count":  solana.scan_count,
            "signal_count": solana.signal_count,
            "routes":      len(solana.routes),
            "paused":      bool(getattr(solana, "paused", False)),
        } if solana is not None else None),
    }


async def _dispatch_action(bot: Any, action: str, params: dict) -> str:
    """Compute one action's result. Fire-and-forget actions return a short
    confirmation string (the real audit trail is the Telegram message they
    send); query actions return the rendered markdown directly. Any
    failure is rendered as an error string rather than raised, so a bad
    request can't 500 the whole endpoint for an unrelated reason."""
    from modules import dashboard_panels as panels

    handlers = getattr(bot, "_handlers", None)
    chat_id = getattr(bot.config, "telegram_chat_id", None)

    if action in _FIRE_AND_FORGET_ACTIONS:
        if handlers is None:
            return "❌ Bot not ready yet — handlers not constructed."
        if action == "toggle_ghost":
            await handlers._cmd_ghost_mode(chat_id)
        elif action == "toggle_mint":
            await handlers._cmd_mint_mode(chat_id)
        elif action == "hunt":
            await handlers._cmd_hunt(chat_id)
        return f"✅ {action} executed — see Telegram for confirmation."

    if action in _QUERY_ACTIONS:
        if action == "wallet_status":
            return await panels.render_wallet_status(bot)
        if action == "gas_health":
            return await panels.render_gas_health(bot)
        if action == "price_check":
            return await panels.render_price_check(bot, params.get("assets", ""))
        if action == "scanner_tokens":
            return await panels.render_scanner_tokens(bot, _auto_hunt_enabled())
        if action == "solana_prices":
            return await panels.render_solana_prices(bot)
        if action == "solana_routes":
            return await panels.render_solana_routes(bot)
        if action == "arbitrage_detail":
            return await panels.render_arbitrage_detail(bot)
        if action == "audit_log":
            return await panels.render_audit_log(bot)
        if action == "speedtest":
            return await panels.render_speedtest(bot)

    return f"❌ Unknown action: {action}"


def _read_recent_log_lines(limit: int = 200) -> str:
    """v18.1 — tail the bot's own rotating log file.

    Operator need: the Hugging Face dashboard runs in DASHBOARD_ONLY mode and
    therefore has no bot of its own, so the only logs it could ever show were
    its own (empty) process — i.e. never the AWS bot actually trading. This
    serves the real thing.

    Reads modules/logger.py's LOG_FILE rather than shelling out to
    journalctl: no subprocess, no sudo, and it works regardless of which
    user the service runs as or whether journald is even in use. Bounded to
    the tail so a multi-megabyte log can't be pulled into memory or pushed
    down the wire in one response."""
    path = os.getenv("LOG_FILE", "").strip()
    if not path:
        return ("LOG_FILE is not set, so the bot is not writing a log file — "
                "only stdout/journald. Set LOG_FILE=bot.log in .env and "
                "restart to enable this view.")
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            # Seek to a bounded window from EOF instead of reading the whole
            # file; 200 lines is far under 256 KiB in this log's format.
            try:
                fh.seek(0, os.SEEK_END)
                fh.seek(max(0, fh.tell() - 256_000))
                fh.readline()  # discard the partial line the seek landed in
            except OSError:
                pass
            lines = fh.readlines()
        tail = [ln.rstrip("\n") for ln in lines[-limit:]]
        return "\n".join(tail) if tail else "(log file is empty)"
    except FileNotFoundError:
        return f"Log file not found at {path} — the bot may not have written to it yet."
    except Exception as exc:
        return f"Could not read log file: {exc}"


async def run_bot_api_server(bot: Any) -> None:
    """Serve GET /status and POST /command until the bot stops. Spawned as
    a background task from bot.py's start() — see this module's own
    docstring for the full endpoint contract."""
    try:
        from aiohttp import web
    except ImportError:
        logger.error(
            "[BotAPI] aiohttp not installed — the bot's HTTP status/"
            "command API will not run. pip install aiohttp."
        )
        return

    token = _bot_api_token()
    if not token:
        logger.warning(
            "[BotAPI] BOT_API_TOKEN is not set — /status will still work "
            "(no secrets in it), but /command will reject every request. "
            "Set BOT_API_TOKEN to let the Hugging Face dashboard control "
            "this bot remotely."
        )

    async def handle_status(request: "web.Request") -> "web.Response":
        try:
            snapshot = await _build_status_snapshot(bot)
            return web.json_response(snapshot)
        except Exception as exc:
            logger.warning("[BotAPI] /status failed: %s", exc)
            return web.json_response({"error": str(exc)}, status=500)

    async def handle_command(request: "web.Request") -> "web.Response":
        configured_token = _bot_api_token()
        if not configured_token:
            return web.json_response(
                {"ok": False, "error": "BOT_API_TOKEN not configured on this bot"},
                status=503,
            )
        auth = request.headers.get("Authorization", "")
        if not hmac.compare_digest(auth, f"Bearer {configured_token}"):
            # v18.2 — feed the failure to modules/http_guard.py, which locks
            # the client out after a handful of attempts. Without this, the
            # only thing bounding a brute-force run against a fund-moving
            # endpoint is how fast the attacker's network is.
            _note_auth(request, ok=False)
            return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
        _note_auth(request, ok=True)

        try:
            body = await request.json()
        except Exception:
            return web.json_response({"ok": False, "error": "invalid JSON body"}, status=400)

        action = (body or {}).get("action")
        params = (body or {}).get("params") or {}
        if not action:
            return web.json_response({"ok": False, "error": "missing action"}, status=400)

        try:
            logger.info("[BotAPI] executing remote command: %s", action)
            result = await _dispatch_action(bot, action, params)
            return web.json_response({"ok": True, "result": result})
        except Exception as exc:
            logger.warning("[BotAPI] action %s failed: %s", action, exc)
            return web.json_response({"ok": False, "error": str(exc)}, status=500)

    async def handle_logs(request: "web.Request") -> "web.Response":
        """Bearer-gated like /command, not open like /status: log lines are
        operational detail (wallet addresses, tx signatures, route sizes),
        which /status deliberately does not expose."""
        configured_token = _bot_api_token()
        if not configured_token:
            return web.json_response(
                {"ok": False, "error": "BOT_API_TOKEN not configured on this bot"},
                status=503,
            )
        if not hmac.compare_digest(
            request.headers.get("Authorization", ""), f"Bearer {configured_token}"
        ):
            _note_auth(request, ok=False)
            return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
        _note_auth(request, ok=True)
        try:
            limit = min(int(request.query.get("limit", 200)), 1000)
        except ValueError:
            limit = 200
        text = await asyncio.to_thread(_read_recent_log_lines, limit)
        return web.json_response({"ok": True, "logs": text})

    # ── Public read-only surface for bot.elghaly.dev (v18.1, 2026-07-29) ──
    #
    # Operator: "a public read-only stats page vs an authenticated control
    # page — you already have this instinct (the /logs 401 earlier was
    # correct behavior, keep that pattern everywhere)."
    #
    # So these four routes are UNAUTHENTICATED BY DESIGN and carry nothing
    # that needs protecting: the page itself is a static string, and both
    # JSON endpoints read only the trade journal (route names, sizes,
    # spreads, on-chain signatures — all of it either already public on
    # Solana or meaningless to an attacker). No wallet address, no RPC URL,
    # no key material, no config, and above all no way to ACT. /command
    # keeps its Bearer token; /logs keeps its 401.
    #
    # That line matters more than it looks: this page is meant to be public
    # at bot.elghaly.dev, and a public page that can move funds is a public
    # page that will move funds.
    async def handle_index(request: "web.Request") -> "web.Response":
        from modules.web_dashboard import INDEX_HTML

        return web.Response(text=INDEX_HTML, content_type="text/html")

    async def handle_api_pnl(request: "web.Request") -> "web.Response":
        from modules.ledger_report import pnl

        # to_thread: these read (tailed) CSVs off disk, and the event loop
        # this shares with the scan loop must never block on I/O.
        return web.json_response(await asyncio.to_thread(pnl))

    async def handle_api_receipts(request: "web.Request") -> "web.Response":
        from modules.ledger_report import receipts

        try:
            # max(1, ...) as well as min(..., 500): Python slices backwards
            # on a negative bound, so `?limit=-1` would have returned
            # nearly every row and defeated the cap entirely. This endpoint
            # is deliberately unauthenticated, so its parameters have to be
            # robust against someone who is not being polite.
            limit = max(1, min(int(request.query.get("limit", 100)), 500))
        except (TypeError, ValueError):
            limit = 100
        rows = await asyncio.to_thread(receipts, limit)
        return web.json_response({
            "receipts": rows,
            "verifiable": True,
            "note": (
                "Each signature is independently checkable on Solana. This "
                "endpoint returns ONLY trades that produced a real on-chain "
                "signature — never estimates, simulations or projections."
            ),
        })

    async def handle_receipts_csv(request: "web.Request") -> "web.Response":
        from modules.ledger_report import receipts_csv

        text = await asyncio.to_thread(receipts_csv)
        return web.Response(
            text=text, content_type="text/csv",
            headers={"Content-Disposition": 'attachment; filename="receipts.csv"'},
        )

    async def handle_api_routes(request: "web.Request") -> "web.Response":
        """v18.2 — route-pruning transparency, public.

        Operator: "expose why a route got pruned (e.g. '0 signals in 500
        evals over 6h') so you can tell a dead route from a temporarily
        quiet one."

        Serves route_scorer.public_table(), which is the internal table
        with every dollar figure removed — see that method for the audit.
        Route names and basis points are safe to publish; loan sizes are
        position information and are not.
        """
        from modules.route_scorer import get_scorer

        scorer = get_scorer()
        try:
            table = await asyncio.to_thread(scorer.public_table)
            status = scorer.status()
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] /api/routes failed: %s", exc)
            return web.json_response({"routes": [], "error": "unavailable"})
        return web.json_response({
            "routes": table,
            "pruning_enabled": status.get("enabled"),
            "routes_pruned": status.get("routes_pruned"),
            "min_active_routes": status.get("min_active_routes"),
            "dead_bps": status.get("dead_bps"),
            "floor_bound": status.get("floor_bound"),
            "probe_every_cycles": status.get("probe_every_cycles"),
            "note": (
                "A pruned route is asleep on a timer, not deleted — it is "
                "re-probed on a fixed cycle and reinstated the moment it "
                "produces. `explain` states the evidence and the observation "
                "window for every route, pruned or not."
            ),
        })

    async def handle_api_engine(request: "web.Request") -> "web.Response":
        """v18.3 — the engine's own instruments, in one payload.

        Operator: "add some more data" — and, in the same breath, "make
        secure just me". Those two travel together: this endpoint carries
        edge-decay measurements, Jito tip state, the trading circuit
        breaker and the drawdown watcher, which is meaningfully more than
        the earlier public payloads exposed.

        Each block is still built by hand rather than dumping the modules'
        full status() dicts, because those are written for /doctor — an
        authenticated, operator-only view — and would carry the signing
        wallet, file paths and threshold config out onto a web page. The
        rule from the public endpoints has not been relaxed just because
        there is a password in front of it now: a password is one
        credential, and a page that leaks a wallet address to whoever
        holds it is still a page that leaks a wallet address.
        """
        out: dict[str, Any] = {}

        try:
            from modules.decay_tracker import get_tracker

            st = get_tracker().status()
            out["decay"] = {
                "recording": st.get("recording"),
                "samples": st.get("samples"),
                "min_samples": st.get("min_samples"),
                "usd_per_sec": st.get("decay_usd_per_sec"),
                "pipeline_median_secs": st.get("pipeline_median_secs"),
                "pipeline_p90_secs": st.get("pipeline_p90_secs"),
                "implied_adder_usd": st.get("implied_adder_usd"),
                "gate_enabled": st.get("gate_enabled"),
            }
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] decay block failed: %s", exc)

        try:
            from modules.trading_guard import get_guard

            st = get_guard().status()
            out["guard"] = {
                "state": st.get("state"),
                "trips": st.get("trips"),
                "attempts": st.get("attempts"),
                "successes": st.get("successes"),
                "blocked": st.get("blocked"),
                "reason": (st.get("trip_reason") or "")[:200],
            }
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] guard block failed: %s", exc)

        try:
            from modules.jito_tip_engine import get_tip_engine

            st = get_tip_engine().status()
            out["jito"] = {
                "gamma": st.get("gamma"),
                "bundles_sent": st.get("bundles_submitted"),
                "bundles_landed": st.get("bundles_landed"),
                "landing_rate": st.get("landing_rate"),
                "priced_out": st.get("priced_out"),
                "tips_paid_sol": st.get("tip_sol_paid"),
                # COUNT, not the list. The block-engine URLs are not secret,
                # but they are infrastructure detail with no reason to be on
                # a web page — and "4 engines" is the whole of what a reader
                # needs from that field.
                "engines": len(st.get("block_engines") or []),
            }
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] jito block failed: %s", exc)

        try:
            from modules.drawdown_alert import get_watcher

            st = get_watcher().status()
            out["drawdown"] = {
                "enabled": st.get("enabled"),
                "halt_on_breach": st.get("halt_on_breach"),
                "thresholds": st.get("thresholds"),
                "latches": st.get("latches"),
            }
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] drawdown block failed: %s", exc)

        try:
            from modules.http_guard import get_guard as get_http_guard

            st = get_http_guard().status()
            out["http"] = {
                "served": st.get("served"),
                "throttled_429": st.get("throttled_429"),
                "shed_503": st.get("shed_503"),
                "auth_lockouts": st.get("auth_lockouts"),
                "locked_clients": st.get("locked_clients"),
            }
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] http block failed: %s", exc)

        return web.json_response(out)

    async def handle_api_nearmiss(request: "web.Request") -> "web.Response":
        """v18.2 — failure-mode logging, public.

        Operator: "logging near-misses — signals that crossed your
        detection threshold but reverted or got beaten to the block —
        since that tells you if you're losing races, not just whether
        you're finding opportunities."

        Public because it is the honest half of the story. A page showing
        only executed trades reads as either "no losses" or "no activity"
        with no way to tell which; publishing the near-misses alongside is
        what makes the receipts section a claim about completeness rather
        than a selection.
        """
        from modules.near_miss import get_log

        near = get_log()
        try:
            limit = max(1, min(int(request.query.get("limit", 25)), 200))
        except (TypeError, ValueError):
            limit = 25
        try:
            summary = await asyncio.to_thread(near.summary)
            recent = await asyncio.to_thread(near.recent, limit)
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] /api/nearmiss failed: %s", exc)
            return web.json_response({"summary": {}, "recent": [], "error": "unavailable"})
        # log_path is a server filesystem path — useful in /doctor, not on
        # a public endpoint, so it is dropped here rather than never
        # collected.
        summary.pop("log_path", None)
        return web.json_response({"summary": summary, "recent": recent})

    async def handle_api_observatory(request: "web.Request") -> "web.Response":
        """v18.5 — the recorded series, not another snapshot.

        Operator's request, in their own framing: a public view of Solana
        tip/MEV conditions that runs continuously, because other bot
        builders want that data and producing it risks no capital.

        Everything here is either published by Jito already or measured
        from public quotes, so there is nothing to withhold — but the
        endpoint-name labels from rpc_latency are deliberately env-var
        names ("SOLANA_RPC_PRIMARY"), never URLs, because the primary RPC
        embeds a provider API key in its path on this deployment. See
        scripts/observe.py, which records them that way for this reason.
        """
        from modules.observatory import viability

        try:
            data = await asyncio.to_thread(viability)
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] /api/observatory failed: %s", exc)
            return web.json_response({"rows": 0, "error": "unavailable"})
        return web.json_response(data)

    async def handle_api_lighthouse(request: "web.Request") -> "web.Response":
        """v18.4 — the verdict that ranks the other panels, public.

        Operator: "improve his web its so ugly."

        The page opened with a table, which meant a reader — including the
        operator on a phone — had to assemble the answer from five sections
        before knowing whether anything was wrong. This serves the same
        five-gate judgement /lighthouse gives in Telegram, so the page can
        lead with the conclusion and keep the evidence underneath it.

        `public=True` drops the `action` field. The findings carry shell
        commands naming scripts and env vars, and operator instructions on
        a web page are exactly what this file's payload rule exists to
        prevent. The diagnosis stays — knowing that the pipeline is slow is
        not a credential — only the runbook line is withheld.
        """
        from modules.lighthouse_report import lighthouse

        try:
            data = await asyncio.to_thread(lighthouse, bot, True)
        except Exception as exc:  # noqa: BLE001
            logger.debug("[BotAPI] /api/lighthouse failed: %s", exc)
            return web.json_response({"stages": [], "verdict": None,
                                      "error": "unavailable"})
        return web.json_response(data)

    # v18.2 — modules/http_guard.py. The server now shares an event loop
    # with the scan loop AND faces the public internet, which makes an
    # enthusiastic scraper indistinguishable from an attack in its effect:
    # scan cadence stretches and the bot stops trading. The middleware caps
    # concurrency, buckets per client, and locks out repeated failed
    # authentication against /command. It fails OPEN on internal error —
    # a broken limiter must not take down the page the operator diagnoses
    # everything else from.
    middlewares = []
    try:
        from modules.http_guard import make_middleware

        middlewares.append(make_middleware())
    except Exception as exc:  # noqa: BLE001
        logger.warning("[BotAPI] HTTP guard unavailable, serving unprotected: %s", exc)

    app = web.Application(middlewares=middlewares)
    app.router.add_get("/", handle_index)
    app.router.add_get("/api/pnl", handle_api_pnl)
    app.router.add_get("/api/receipts", handle_api_receipts)
    app.router.add_get("/api/routes", handle_api_routes)
    app.router.add_get("/api/nearmiss", handle_api_nearmiss)
    app.router.add_get("/api/engine", handle_api_engine)
    app.router.add_get("/api/lighthouse", handle_api_lighthouse)
    app.router.add_get("/api/observatory", handle_api_observatory)
    app.router.add_get("/receipts.csv", handle_receipts_csv)
    app.router.add_get("/status", handle_status)
    app.router.add_post("/command", handle_command)
    app.router.add_get("/logs", handle_logs)

    host, port = _bot_api_host(), _bot_api_port()
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, host, port)
    await site.start()
    logger.info("🌐 Bot API listening on %s:%d (/status, /command)", host, port)

    try:
        while getattr(bot, "_running", True):
            await asyncio.sleep(1)
    finally:
        await runner.cleanup()