modules/command_handlers.py

"""
modules/command_handlers.py — Garden Angel Command Handlers v17.36
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
v17.36 (2026-07-28, operator review) — vs v17.28:

  FEATURE — six commands for the Solana engine's new subsystems. Kept
    separate from /doctor rather than folded into it: /doctor answers
    "can it trade right now", these answer "why is it behaving like
    this", and one message carrying both is one nobody reads on a phone.

      /routes   per-route win rate mined from data/trade_journal.csv, and
                which routes are currently pruned WITH the numbers that
                pruned them. The answer to "why did SOL->BONK stop
                appearing" is a line of arithmetic, not a shrug.
      /whatif   replay a proposed config change against recorded history
                BEFORE redeploying. `/whatif haircut=15` reproduces the
                v1.47 disaster (USDC->SOL->USDC: 34 signals -> 0) from
                real rows in about a second.
      /jito     the dynamic tip auction: live γ, tip floor, landing rate,
                bundles landed vs priced out, engine list.
      /venues   live probe of the direct-DEX fallback adapters, run FROM
                the box that would be calling them — so "did Orca move
                its endpoint" is one command, not a debugging session.
      /resume   clear the trading circuit breaker after an auto-pause.
      /pumpfun  pump.fun verification API status + credential probe.

  FEATURE — /doctor gained the three checks that most often explain a
    silent bot: the trading circuit breaker's state (as a ⛔ problem when
    open, since it is the single likeliest reason a correctly-configured
    bot is not executing), key custody (a plaintext key is a ⚠️ warning,
    a key/wallet mismatch is a ⛔ problem), and the effective vs
    configured profit floor when volatility has raised it.

v17.28 (2026-07-27, operator request — "add doctor command at telegram to
know what's the problem", modeled on their separate Rust bot's /doctor whose
one-line verdict "✅ no problems found — configuration can trade." ended a
whole class of why-no-trade back-and-forth) — vs v17.27:

  FEATURE — /doctor: plain-language diagnosis of the Solana leg. Splits
    findings into ⛔ problems (things that BLOCK trading: execution
    blockers from SolanaArbEngine._execution_blockers(), a stale scan
    loop, an empty fee wallet, a mis-set leg-2 haircut big enough to eat
    every edge — the exact config bug fixed 2026-07-27), ⚠️ warnings
    (rate-limit cooling, warm-up, unreadable balance), and ℹ️ state (min
    profit, drift buffer, learned flash fee, last signal's real outcome).
    Read-only by construction: every check reads state that already
    exists, plus at most ONE best-effort getBalance RPC — zero Jupiter
    calls, so running it never spends rate-limit budget or perturbs an
    in-flight scan. Ends with the Rust bot's exact verdict line when
    nothing is wrong, so the operator learns to trust one sentence.

v17.27 (2026-07-18, operator forwarded more "🤖" replies with the same shape
as v17.25's — fabricated numbers, invented DEX-fee comparisons, a made-up
Camelot/SushiSwap triangle strategy — then asked to "fix his mind ... make
him small agent") — vs v17.25:

  FEATURE — Solana gets the exact same small, grounded, operator-approved
    agent the ARB /ai_idea/ai_do → /approve flow already is (v17.25 fixed
    what /ai's free-form context KNOWS; this gives Solana an action
    surface with the same safety shape as ARB's, not just better facts):

      /ai_idea_sol — new command, mirrors _cmd_ai_idea() exactly: calls
        qwen_client.propose_solana_route() (schema-limited to constants.
        SOLANA_TOKENS symbols), grounds both symbols via SolanaArbEngine.
        verify_solana_mint() before ever showing the operator anything,
        stores the proposal as a chain="SOL"-tagged self._pending_ai_pair.

      /ai_do sol <request> — a leading "sol" token routes to a Solana-
        specific action set (pause/resume/set_min_profit/add_route/
        remove_route — no set_loan, no add_pair's 0x-address shape, see
        qwen_client.py's matching v1.22 entry) built from the Solana leg's
        own real numbers (routes, floor, paused, last near-miss), instead
        of always defaulting to ARB.

      _cmd_approve()/_cmd_reject() branch on the pending pair's chain tag:
        SOL calls SolanaArbEngine.add_route() and persists to the new
        ai_solana_routes.json (bot.py loads/re-adds these at every startup,
        mirroring ai_pairs.json's own _ai_approved_pairs() exactly) instead
        of Scanner.add_pair()/ai_pairs.json.

      _apply_ai_action() branches the same way for pause/resume/
        set_min_profit/add_route/remove_route, targeting self._solana
        instead of self._scanner when the action's chain tag says SOL.
        New SolanaArbEngine.paused flag (solana_arb.py's own matching
        v1.44 entry) gives Solana a real pause switch for the first time —
        it never had one before this.

    None of this changes how /ai (ask()) itself answers open questions —
    that free-form path has no schema to ground against by design, and
    stays exactly as risky as any open Q&A always is. What's new is a
    second, narrow, verified path standing next to it for anything that
    should actually change what the bot watches or does.

v17.25 (2026-07-17, operator forwarded "🤖" AI replies confidently citing a
"$500k Solana loan"/"$250 Solana fee"/"$1.50 floor" — numbers that exist
NOWHERE in this deployment's Solana config (real probe $10k, real Kamino fee
0 bps, real floor $2 or $1 in mint mode); the ARB-only loan_amount/
loan_fee_pct WERE in /ai's context with no chain label, and the Solana leg
had no numbers there at all, so the model reused ARB figures for Solana
claims and invented the rest) — vs v17.24:

  FIX (grounding) — _cmd_ai()'s context now includes the Solana leg's REAL
    numbers whenever self._solana is attached: min_profit floor, probe
    size, the real-send-is-a-tiny-fraction-of-probe caveat (the exact
    mechanism behind every "real size too thin" outcome the operator sees
    on the dashboard), flash fee bps, and the latest scan's best route
    with its actual gap to floor (SolanaOpportunity.near_miss_line — the
    same grounded line the near-miss Telegram report already uses). The
    ARB loan/fee entries are now explicitly labeled ARBITRUM-ONLY so they
    can't be silently reused as Solana figures again. Pairs with
    qwen_client.py's own v1.19 (venue-facts + say-when-not-in-context
    prompt rules, and the matching context-cap raise so this longer
    context isn't truncated).

v17.24 (2026-07-17, operator report — "we need to make the ai like agent
suggestion and fix and help the bot it's still not work", screenshot
showing two consecutive /ai_idea calls both replying "No usable suggestion
this time" with no explanation) — vs v17.23:

  FIX (root cause) — _cmd_ai_idea() accepted or rejected propose_pair()'s
    candidate address using only a 0x…40-hex REGEX check (inside
    qwen_client.py) — never checked whether a contract actually exists
    there. A single LLM call either hit that regex or didn't, with no
    retry and no visibility into which validation step failed, so a bad
    call meant the operator had to keep manually resending /ai_idea with
    no better odds each time and no idea why.

  NEW (grounding) — now retries up to _AI_IDEA_MAX_ATTEMPTS=3 times
    server-side per /ai_idea, and each candidate is cross-checked against
    the live chain via scanner.py's new verify_erc20_token() (v18.46):
    confirms a contract actually exists at the address and reads its REAL
    symbol()/decimals(), overriding the AI's guessed decimals and flagging
    a symbol mismatch if the on-chain symbol doesn't match what the AI
    claimed. Failure messages now show the actual reason (from
    qwen_client.py's own v1.18 last_error fix, or this file's own
    verify_erc20_token rejection reason) instead of one generic line.

v17.23 (2026-07-17, operator report — a real /ai answer cited "flash-loan
fees (typically 0.09% on Arbitrum, or $450)" on a $500k loan; this
deployment's REAL configured rate, per the Scanner's own startup log, is
0.05% ($250)) — vs v17.22:

  FIX (grounding) — _cmd_ai()'s context string never included the
    scanner's own loan_fee_pct either, same gap loan_amount had before
    v17.22's fix, so the model had no real fee rate to check its own
    claim against and fell back to a generic "typical Aave-style fee"
    figure. Added `flash-loan fee: X% (this deployment's REAL configured
    rate...)` to the context. Pairs with qwen_client.py's own v1.17 entry,
    which adds the matching general-purpose "prefer given real numbers
    over memorized typical figures" prompt guardrail.

v17.22 (2026-07-17, operator report — a real /ai answer recommended
lowering the profit floor to catch a "0.15-0.25% spread," a claim that
doesn't reconcile with the bot's actual loan size) — vs v17.21:

  FIX (grounding) — _cmd_ai()'s context string never included the
    scanner's own loan_amount, so the model had no way to check a spread/
    dollar claim it made against the real trade size before suggesting a
    floor change. Added `loan size per trade: $X` to the context. Pairs
    with qwen_client.py's own v1.14 entry, which adds the matching prompt
    guardrail that actually uses this number.

v17.21 (2026-07-15) — vs v17.20:

  FIX (diagnosability) — companion to qwen_client.py v1.9. `/ai <question>`
    replied with the same "🤖 The AI didn't answer — try again in a
    minute." on every kind of failure (bad API key, wrong model name,
    timeout, rate limit) with no way for the operator to tell which.
    `_cmd_ai` now appends `self._qwen.last_error` (the new v1.9 property)
    to that message when set, so the next occurrence names the real cause
    instead of a dead end — same fix shape as telegram_client.py's earlier
    403-description fix this session.

v17.20 (2026-07-11) — vs v17.19:

  FIX (root cause) — Mint Mode defaulted ON. __init__ set
    self._mint_mode = True at construction, and effective_min_profit
    (used by BOTH _run_hunt's manual /hunt path and run_autonomous_scan's
    24/7 loop — the only two callers of scanner.scan()) halves
    self._min_profit whenever mint_mode is True. With min_profit=$4.35
    (MIN_PROFIT_USD), that's 4.35 * 0.5 = $2.175 — the exact "$2.17"
    floor scanner.py v18.22-v18.25 traced across multiple sessions
    without being able to find the source, since Scanner's own
    "Constructed with min_profit=$4.35" diagnostic only ever reflects the
    static constructor baseline, never the per-scan override actually
    used. Confirmed via scanner.py's scan()-level override-mismatch
    warning (v18.22): it logs whenever the caller's override differs
    from baseline by >$0.01, and its total absence from a full log
    window while mint_mode defaulted True was the tell that this file,
    not scanner.py/bot.py/config.py, was the source.

    Root cause of root cause: _cmd_force_reset also re-set
    self._mint_mode = True under a comment claiming it restored "safe
    defaults" — so even an operator who noticed mint mode was on and
    force-reset the bot would have it silently re-enabled.

    Fix: both defaults changed to False. The bot now starts — and
    force-resets — with the full configured $4.35 floor unless mint
    mode is explicitly turned on via /mint_mode. The toggle itself is
    unchanged; this only changes what it defaults to.

v17.19 (2026-07-10) — vs v17.18:

  REMOVE — /quote and /execute. These were manual, hand-typed on-chain
    trade commands ported from command_handlers_5.py in the v17.16 merge.
    They're now strictly worse than the automated path: LocalExecutor
    (modules/local_executor.py) already does exact constant-product
    on-chain quoting, fee-aware venue selection across every DEX
    combination, and a live pre-flight profitability check before ever
    sending a transaction — none of which /quote or /execute replicate.
    /execute in particular let an operator hand-craft minIntermediateOut/
    minFinalOut and fire a real flash-loan trade with no automated
    profitability gate at all beyond what the human typed. Keeping a
    manual bypass around a safer automated path is a pure liability, not
    a feature — removed both handlers, their dispatcher branches, their
    HELP_TEXT lines, and _EXECUTE_SLIPPAGE_BUFFER_PCT (only ever consumed
    by the now-removed /quote). "quote"/"execute" also dropped from
    _ON_CHAIN_COMMANDS. /withdraw, /withdraweth, /chainstatus, /owner,
    /sweep are unaffected — those are distinct utilities (fund recovery,
    read-only status, wallet sweep), not trade-execution bypasses.

  FIX — /payout no longer attempts the oracle POST. self._oracle_url
    pointed at the Cloudflare Worker's /payout route, which no longer
    exists on the now-passive-relay Worker (confirmed dead — the sibling
    /execute route already returns {"error":"route not found"} in
    production). Every /payout call was burning up to 20s waiting on a
    request that can only ever fail, then falling through to the report-
    export path anyway. Removed the oracle-POST block entirely; /payout
    now goes straight to the working .js report export. oracle_responded
    is hardcoded False in the report (nothing was attempted to respond).

v17.18 (2026-07-07) — vs v17.17:

  FIX — /payout's oracle path (_cmd_payout) was treating HTTP 200 alone
    as a successful payout, with no check that a tx_hash was present or
    that the Worker's response indicated an actual on-chain confirmation.
    This is the same class of gap oracle.py's execute_trade() already
    closed for Scanner's BUY signals (status == "confirmed" AND a
    non-empty tx_hash — anything else is a failed execution), just
    reopened on a different code path (/payout, not /execute) that never
    reused that contract. Fixed: a 200 response is now only reported as
    "Payout Successful" when tx_hash is non-empty AND status (if present)
    is "confirmed". A 200 that fails that check now falls through to the
    same unconfirmed/report-export path a non-200 response already took,
    instead of claiming success on an unvalidated body.

v17.17 (2026-07-06) — vs v17.16:

  NEW — /payoutreset [eth] [confirm] [full] command.

  Root cause: /reset and /force_reset only ever cleared in-memory
  _stats/_recent_results/price cache. Neither one — nor any other
  command in this file — ever called PayoutManager.reset(). That meant
  the actual accounting ledger (gross_profit/net_profit/buy_count/
  scan_count, the "Payout Ledger — BSC (source of truth)" block in
  /status and /health) had NO command capable of clearing it, and could
  carry stale pre-oracle-gating-fix numbers indefinitely with no way to
  zero them from Telegram. Confirmed in production: ledger showed
  net_profit=$543.10 with buy_count=0 and scan_count=4 — numbers that
  predate the oracle.execute_trade() confirmation gate landing in
  scanner.py's _notify_payout(), never reset since.

  /payoutreset defaults to BSC, accepts 'eth' for the ETH ledger (if
  configured), previews current state and requires an explicit
  'confirm' arg before actually resetting (destructive to accounting
  history, though never to on-chain funds/wallet balances), and accepts
  'full' to also clear total_swept_wei/anomaly_strikes via
  PayoutManager.reset()'s existing full_reset param.

v17.16 (Architecture Merge) — vs v17.15:

  This dispatcher was kept as the core framework over the parallel
  python-telegram-bot prototype (command_handlers_5.py) — lower overhead,
  and everything else in this bot (scanner, price client, netdiag) is
  already wired to the custom httpx-based TelegramClient, not to PTB's
  Application runtime. Rewriting the whole bot onto PTB to gain 6 commands
  wasn't a good trade.

  NEW — /chainstatus, /owner, /quote, /execute, /withdraw, /withdraweth.
    On-chain FlashArbitrageV2 execution, ported from command_handlers_5.py.
    The web3.py logic itself moved into its own modules/contract_manager.py
    (framework-agnostic, no telegram imports) rather than being pasted
    inline here — matches how scanner/price_client/payout_manager are
    already split out. ContractManager's calls are synchronous web3 I/O;
    every call site here wraps them in asyncio.to_thread so a slow RPC
    node can't stall the dispatcher, the same fix v17.14 already applied
    to /hunt for the same underlying reason (blocking call inline in an
    async handler blocks every other command from being seen).

  NEW — On-chain command allowlist (_ON_CHAIN_ALLOWED_CHAT_IDS /
    TELEGRAM_ALLOWED_CHAT_IDS). command_handlers_5.py gated every single
    command behind a chat_id allowlist; this dispatcher previously gated
    none of them. Rather than either extreme, only the 6 new on-chain
    commands are gated here — they're the ones that can move funds or
    expose contract internals, and "any chat that can message this bot"
    was never an acceptable authorization boundary for /execute or
    /withdraw regardless of what else got merged first. The rest of the
    bot's commands remain ungated, same as v17.15 — that's a separate,
    deliberately-deferred follow-up, not an oversight in this merge.

v17.15 ADDITION (Hardening Phase 1 — Visibility) — vs v17.14:

  NEW — /netdiag command. modules/netdiag.py was already wired into
    bot.py's startup path, but only fires automatically when a startup
    check fails — by design, so a healthy boot isn't slowed down by a
    ~5-host IPv4/IPv6 probe. That left no way to check network health
    on demand once the bot is already running ("flying blind" between
    failures). /netdiag calls the same diagnose_network() function and
    renders its per-host IPv4/IPv6 verdict as a Telegram message instead
    of (only) a log line, so you can pull a live network read any time
    from your phone without needing log/SSH access in the moment.
    Degrades cleanly (clear message, no crash) if modules/netdiag.py
    isn't deployed in a given environment.

v17.14 FIXES (vs v17.13) — Incident Report §3.3:

  FIX — /hunt no longer runs inline inside the dispatcher.
    Previously `handle()` awaited the entire scan (oracle/OKX/CoinGecko/
    Binance round-trips + the scanner itself) before returning. If
    bot.py's update loop awaits handle() per-update (the usual pattern),
    a single /hunt could block that loop from seeing the next Telegram
    update for the full scan duration — read by users as "the bot drops
    commands / looks unresponsive."

    Now /hunt:
      1. Returns immediately after an ack message if a scan can be
         started (or a "scan already running" message if not).
      2. Runs the actual scan as a tracked background asyncio.Task
         (self._spawn), so handle() — and whatever loop calls it — is
         free to keep processing updates while the scan runs.
      3. Is guarded by self.scan_lock, a public asyncio.Lock. Wire
         bot.py's scheduled auto-scan loop to `async with handlers.
         scan_lock:` around its own scanner.scan() call so a manual
         /hunt and the automatic loop can never run concurrently and
         double up on oracle/OKX/CoinGecko/Binance request volume.
      4. On shutdown, bot.py should `await handlers.aclose()` to let
         any in-flight background scan finish instead of being dropped.

  NOTE (2026-06-30 reconstruction): lines 198-518 of this file were
  lost to a "< truncated lines 198-518 >" placeholder literally being
  saved into the file by a previous paste — this restores the
  dispatcher (`handle()`) and the command handlers it calls into.
  _cmd_hunt() and _cmd_price() probe for the most likely Scanner /
  PriceClient method names defensively (getattr with fallbacks) since
  those two modules weren't available to verify exact signatures —
  if a method name is wrong, the bot still starts and every other
  command still works; only that one command replies with a clear
  error instead of crashing.

  UNCHANGED from v17.13:
    /price, /ping, /audit, /arbitrage, /flashloan, /ghost_mode,
    /mint_mode, /debug, /circuit, /reset, /force_reset, /health,
    /payout, SPINNER FIX, rate-limiting, ring buffer,
    BUTTON_TO_COMMAND mapping.

v17.13 FIXES (kept):
  FIX 1 — /ghost_mode real implementation (toggles self._ghost_mode).
  FIX 2 — /mint_mode real implementation (toggles self._mint_mode).
  FIX 3 — /payout calls oracle first, falls back to report export.
  FIX 4 — /status shows Ghost and Mint mode state.
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
"""

from __future__ import annotations

import asyncio
import hmac
import json
import logging
import os
import time
from datetime import datetime, timezone
from typing import Any, Optional, TYPE_CHECKING

from settings import _env_bool
from .telegram_client import TelegramClient, TelegramUpdate
from .price_client import PriceClient
from .scanner import Scanner
from .contract_manager import get_contract_manager, ContractLogicError

if TYPE_CHECKING:
    from .payout_manager import PayoutManager

# Optional — only present once modules/netdiag.py has been deployed.
# Mirrors the same try/except pattern bot.py uses, so /netdiag degrades
# to a clear "not available" message instead of crashing the dispatcher
# in any environment where the module hasn't been added yet.
try:
    from .netdiag import diagnose_network
except ImportError:
    diagnose_network = None  # type: ignore[assignment]

logger = logging.getLogger(__name__)

_BOT_START_TIME = time.time()


def _arb_scan_enabled() -> bool:
    """ARB_SCAN_ENABLED=false retires the EVM (Arbitrum) leg from every
    manual scan surface — /hunt here and the dashboard button (app.py's
    do_hunt), plus bot.py's autonomous loop. Operator decision 2026-07-21
    ("remove arb if not have hope"): at the effective loan sizes the
    liquidity floor allows (~$300-400), ARB's flash fee + gas exceed any
    realistically available spread, so every scan is structural HOLD
    noise. Mirrors SOLANA_SCAN_ENABLED's parsing (default ON; explicit
    0/false/no/off disables). Read per call, not cached, so a Space
    variable flip takes effect on restart without a code change.

    v18.0 — ARB_TRADING_ENABLED (config.py's arb_trading_enabled, default
    FALSE) is also checked here: either flag being off disables ARB. This
    file has no cfg object threaded in (env-only, same as
    SOLANA_SCAN_ENABLED above), so it's read directly, same pattern as
    ARB_SCAN_ENABLED. bot.py's self._active_chains already empties
    Scanner's own ARB pairs list when arb_trading_enabled is false — this
    is the matching gate for the manual/dashboard reporting surfaces this
    function controls, so their messaging doesn't claim ARB is scanning
    when Scanner actually has nothing to scan.
    """
    arb_trading_enabled = os.environ.get("ARB_TRADING_ENABLED", "false").strip().lower() in (
        "1", "true", "yes", "on",
    )
    if not arb_trading_enabled:
        return False
    return os.environ.get("ARB_SCAN_ENABLED", "true").strip().lower() not in (
        "0", "false", "no", "off",
    )

_stats: dict[str, Any] = {
    "scans_total":    0,
    "scans_buy":      0,
    "commands_total": 0,
    "errors_total":   0,
    "last_scan_ts":   None,
    "last_signal":    "N/A",
}

_RECENT_CAP = 20
_recent_results: list[dict[str, Any]] = []

_PAYOUT_MAX_ATTEMPTS = 5
_PAYOUT_WINDOW_SECS  = 600.0
_payout_attempts: dict[int, list[float]] = {}

# ── On-chain command access control (v17.16 merge) ──────────────────────────
# /chainstatus, /owner, /quote, /execute, /withdraw, /withdraweth are new in
# this merge and move real funds (execute/withdraw/withdraweth) or expose
# contract internals. Unlike the rest of this dispatcher, they are gated by
# an explicit chat_id allowlist — ported from command_handlers_5.py's
# @restricted decorator — because "anyone who can message this bot" is not
# an acceptable authorization boundary for a command that can drain the
# contract. This does NOT (yet) apply to the older commands below; that's a
# separate follow-up, not something this merge silently expands or narrows.

def _load_allowed_chat_ids() -> frozenset[int]:
    raw = os.environ.get("TELEGRAM_ALLOWED_CHAT_IDS", "")
    ids: set[int] = set()
    for chunk in raw.split(","):
        chunk = chunk.strip()
        if not chunk:
            continue
        try:
            ids.add(int(chunk))
        except ValueError:
            logger.warning("Ignoring non-integer entry in TELEGRAM_ALLOWED_CHAT_IDS: %r", chunk)
    if not ids:
        # v1.93 — SEVERITY MATCHED TO CONSEQUENCE.
        #
        # This was a WARNING on every single startup of a deployment where
        # it describes nothing wrong. It gates exactly two commands —
        # /chainstatus and /owner, both from the Arbitrum era — and this
        # box runs the Solana leg. The operator's instruction is "dont want
        # see any warn", and a line that is true, unactionable and repeated
        # forever is precisely the kind that teaches people to skim past
        # the warnings that DO matter.
        #
        # It stays a WARNING in the one case where it means the bot is
        # unusable: nothing configured anywhere, so no chat can run
        # anything. The environment is read directly rather than via
        # _OWNER_CHAT_IDS because that is defined further down this file
        # and does not exist yet at import time.
        has_any_chat = any(
            os.environ.get(name, "").strip()
            for name in ("TELEGRAM_OWNER_CHAT_IDS", "TELEGRAM_CHAT_ID")
        )
        if has_any_chat:
            logger.info(
                "[Auth] TELEGRAM_ALLOWED_CHAT_IDS is unset — /chainstatus "
                "and /owner (Arbitrum-era, on-chain) will refuse every chat. "
                "Every other command is authorised normally through "
                "TELEGRAM_CHAT_ID. Set it only if you want those two back."
            )
        else:
            logger.warning(
                "TELEGRAM_ALLOWED_CHAT_IDS is empty — /chainstatus and /owner "
                "will reject every chat until it's set."
            )
    return frozenset(ids)


_ON_CHAIN_ALLOWED_CHAT_IDS = _load_allowed_chat_ids()
# v17.21 — withdraw/withdraweth/sweep removed from Telegram entirely
# (dashboard-only now); the two read-only commands stay allowlisted.
_ON_CHAIN_COMMANDS = frozenset({"chainstatus", "owner"})


# ── v17.40 — the OWNER gate (2026-07-29) ─────────────────────────────────
#
# Operator: "make secure just me don't want anyone talk with the bot."
#
# They were right to ask, and the gap was worse than it looks. The
# allowlist above covers exactly two commands — `chainstatus` and `owner`
# — out of thirty-nine. The comment at line 450 even says so ("This does
# NOT (yet) apply to the older commands below; that's a separate
# follow-up"). That follow-up is this.
#
# A Telegram bot answers ANYONE who finds its @handle. Until now, a
# stranger messaging @Garden_Angel_Bot could run `/doctor` (engine state,
# wallet balance, funding address), `/routes` and `/hunt` (the live
# strategy and its numbers), `/speedtest` (spends this account's Jupiter
# rate-limit budget), and `/ghost_mode` and `/mint_mode`, which CHANGE HOW
# THE BOT TRADES. None of those needed a credential.
#
# So the gate is now global and default-deny: every command, checked
# before dispatch.
#
# It resolves from three sources in order, and the fallback chain is the
# part worth getting right — a lockdown that locks the operator out is
# just an outage:
#
#   1. TELEGRAM_OWNER_CHAT_IDS   explicit, comma-separated
#   2. TELEGRAM_ALLOWED_CHAT_IDS the existing on-chain allowlist
#   3. TELEGRAM_CHAT_ID          where the bot already sends every report
#
# (3) is the safety net. It is already mandatory for the bot to function
# at all, so it can never be unset on a working deployment, which means
# this gate can never produce a bot that answers nobody.
#
# The one way to end up with an open bot is to set TELEGRAM_OWNER_OPEN=true
# deliberately. It logs an error every startup for as long as it is on.
def _load_owner_chat_ids() -> frozenset[int]:
    ids: set[int] = set()
    for source in ("TELEGRAM_OWNER_CHAT_IDS", "TELEGRAM_ALLOWED_CHAT_IDS", "TELEGRAM_CHAT_ID"):
        raw = os.environ.get(source, "")
        for chunk in raw.split(","):
            chunk = chunk.strip()
            if not chunk:
                continue
            try:
                ids.add(int(chunk))
            except ValueError:
                logger.warning("Ignoring non-integer entry in %s: %r", source, chunk)
        if ids:
            logger.info(
                "[Auth] command gate active — %d chat(s) authorised, from %s. "
                "Every other chat is refused.", len(ids), source,
            )
            return frozenset(ids)
    return frozenset()


_OWNER_CHAT_IDS = _load_owner_chat_ids()
_OWNER_GATE_OPEN = os.environ.get("TELEGRAM_OWNER_OPEN", "").strip().lower() in (
    "1", "true", "yes", "on",
)

# Answered for anyone, because refusing them leaks more than answering
# does. A stranger who gets silence learns the bot is alive and gated; a
# stranger who gets "this bot is private" learns the same thing and stops.
# Neither reveals a number, a route, or a balance.
_PUBLIC_COMMANDS = frozenset({"start", "help"})

if _OWNER_GATE_OPEN:
    logger.error(
        "[Auth] TELEGRAM_OWNER_OPEN=true — ANY Telegram user who finds this "
        "bot can run /doctor, /routes, /hunt, /ghost_mode and /mint_mode. "
        "Unset it unless you genuinely mean the bot to be public."
    )
elif not _OWNER_CHAT_IDS:
    logger.error(
        "[Auth] no TELEGRAM_OWNER_CHAT_IDS, TELEGRAM_ALLOWED_CHAT_IDS or "
        "TELEGRAM_CHAT_ID is set — the command gate has nobody to authorise "
        "and will refuse EVERY chat, including yours. Set TELEGRAM_CHAT_ID."
    )

# ── Feature request: Payout TxHash Notification — block explorer links ──────
# FIX (2026-07-15, operator went live — DRY_RUN=false) — "ARB"/"ARBITRUM" was
# missing entirely, even though PAYOUT_CHAIN defaults to "ARB" (config.py)
# post-migration. Every real Arbitrum tx hash was falling through to the
# "ETH" default below and linking to etherscan.io, where an Arbitrum tx can
# never be found — exactly the moment an operator most needs a working link
# to verify their own real trade/sweep.
_EXPLORER_BASE: dict[str, str] = {
    "ARB":      "https://arbiscan.io/tx/",
    "ARBITRUM": "https://arbiscan.io/tx/",
    "ETH":      "https://etherscan.io/tx/",
    "ETHEREUM": "https://etherscan.io/tx/",
    "BSC":      "https://bscscan.com/tx/",
    "BNB":      "https://bscscan.com/tx/",
}


def explorer_link(chain: str, tx_hash: str) -> str:
    """Markdown-link a TxHash to the right block explorer for the chain."""
    if not tx_hash:
        return "N/A"
    # Default fallback is ARB, not ETH — this deployment's live chain.
    base = _EXPLORER_BASE.get((chain or "ARB").upper(), _EXPLORER_BASE["ARB"])
    short = f"{tx_hash[:10]}…{tx_hash[-6:]}" if len(tx_hash) > 18 else tx_hash
    return f"[{short}]({base}{tx_hash})"

# v17.22 (2026-07-15, "one list, easy, be sure all working") — every
# practical no-argument command is now a button here, mirroring
# telegram_client.py's _KEYBOARD_ROWS exactly (see that constant's own
# comment for the full reasoning). Only /ai <question> and /ai_do
# <request> are left out — both need free-text input a button can't
# supply.
BUTTON_TO_COMMAND: dict[str, str] = {
    "Status":       "status",
    "Hunt":         "hunt",
    "Price":        "price",
    "Arbitrage":    "arbitrage",
    "Audit":        "audit",
    "Flashloan":    "flashloan",
    "Gas":          "gas",
    "Chain Status": "chainstatus",
    "Owner":        "owner",
    "Health Check": "health",
    "Debug":        "debug",
    "Circuit":      "circuit",
    "Net Diag":     "netdiag",
    "Speed Test":   "speedtest",
    "Ghost Mode":   "ghost_mode",
    "Mint Mode":    "mint_mode",
    "Ping":         "ping",
    "Reset":        "reset",
    "Force Clear":  "force_reset",
    "AI Chat":      "ai",
    "AI Idea":      "ai_idea",
    "AI Idea Sol":  "ai_idea_sol",
    "Approve":      "approve",
    "Reject":       "reject",
}

HELP_TEXT = (
    "🪬🧿 *Garden Angel v17.28 — Commands*\n\n"
    "_Every command below has a tap button on the Command Center menu "
    "(send /start or /help to see it) — the only one that needs you to "
    "type something is `/ai_do`, since it takes a real request as its "
    "argument. `/ai` works both ways: tap it for the learned model, or "
    "type a question after it._\n\n"
    "*AI Copilot*\n"
    "`/ai` — what the bot has LEARNED: clearing rate by hour of day, "
    "and its landing record. All measured, nothing generated\n"
    "`/ai <question>` — talk to the AI about the bot/market\n"
    "`/ai_idea` — AI proposes ONE new Arbitrum pair to watch\n"
    "`/ai_idea_sol` — AI proposes ONE new Solana route to watch "
    "(base->mid, verified on-chain before you ever see it)\n"
    "`/ai_do <request>` — AI turns a plain-language request into ONE "
    "control action (set loan/min-profit, pause/resume, add/remove pair)\n"
    "`/ai_do sol <request>` — same, for the Solana leg (min-profit, "
    "pause/resume, add/remove route — no loan-size knob, Solana has none)\n"
    "`/approve` / `/reject` — apply (or discard) the AI's pending "
    "idea OR action — nothing changes until you approve\n\n"
    "*Market*\n"
    "`/hunt`  — On-demand arbitrage scan\n"
    "`/price` — Live prices (WETH, BNB, BTC)\n"
    "`/arbitrage` — Last scan snapshot\n"
    "`/flashloan` — Cost-model assumptions\n"
    "`/gas` — Last scan's gas estimate\n\n"
    "*On-Chain* _(allowlisted chats only; read-only)_\n"
    "`/chainstatus` — Contract owner, Aave pool, ETH balance\n"
    "`/owner` — Contract owner address\n"
    "_Payout, sweep and withdraw moved to the Hugging Face dashboard's "
    "Payout tab (operator request, 2026-07-12)._\n\n"
    "*Modes*\n"
    "`/ghost_mode` — Toggle ghost mode (scan but don't execute)\n"
    "`/mint_mode`  — Toggle aggressive threshold mode\n\n"
    "*Start here* _(v17.43)_\n"
    "`/why` — why nothing traded, ranked, ending in what to fix first. "
    "This is the one to read when something looks wrong.\n"
    "`/tune` — what to change, what it would have done (replayed against "
    "your own history), one tap to apply. `/tune undo` reverts.\n"
    "`/pipeline` — where the time between signal and wire goes, stage by "
    "stage\n"
    "`/gates` — which of OUR OWN rules stopped a trade, and what they "
    "were holding\n\n"
    "*Solana engine* _(v17.36)_\n"
    "`/routes` — per-route win rate; which pairs are pruned and why\n"
    "`/whatif floor=1.50` — replay a config change against real history\n"
    "`/jito` — dynamic tip auction: γ, tip floor, landing rate\n"
    "`/venues` — probe the direct Orca/Raydium/Meteora fallback\n"
    "`/resume` — clear the trading circuit breaker after a pause\n"
    "`/pumpfun` — pump.fun verification API status\n"
    "`/decay` — how much edge dies between detection and send\n"
    "`/nearmiss` — why signals never became trades (lost vs blocked)\n"
    "`/drawdown` — drawdown, loss streak, and the alert thresholds\n"
    "`/capture` — of the signals found, what share became trades\n\n"
    "*System*\n"
    "`/doctor` — plain-language diagnosis: can it trade, and if not, why\n"
    "`/status` `/health` `/debug` `/circuit`\n"
    "`/netdiag` — IPv4/IPv6 network diagnostic\n"
    "`/speedtest` — Solana RPC/WS + Jupiter latency diagnostic\n"
    "`/audit` — Recent scan history\n"
    "`/reset` `/force_reset` — Clear stats/cache\n"
    "`/ping`   — Latency check\n"
    "`/testreport` — Send a test message to the report channel\n"
    "`/testlog` — Send a test message to the payout/log channel\n\n"
    "_Tap a button below or type a command._"
)


class CommandHandlers:

    def __init__(
        self,
        telegram:       TelegramClient,
        scanner:        Scanner,
        price_client:   PriceClient,
        payout_manager: Optional["PayoutManager"] = None,
        payout_manager_eth: Optional["PayoutManager"] = None,
        dry_run:        bool  = False,
        min_profit:     float = 100.0,
        scan_timeout:   float = 20.0,
        oracle_url:     str   = "",       # kept for constructor compat; unused since v17.19 (/payout no longer POSTs to it — see module changelog)
        qwen:           Any | None = None,   # v17.21 — AI copilot (/ai, /ai_idea)
        ai_pairs_path:  str   = "",         # v17.21 — where /approve persists pairs
        ai_solana_routes_path: str = "",    # v17.27 — where /approve persists Solana routes (/ai_idea_sol)
        report_channel: Any | None = None,  # v18.42 — /testreport; None if TELEGRAM_REPORT_CHAT_ID unset
    ) -> None:
        self._tg           = telegram
        self._scanner       = scanner
        self._price         = price_client
        self._payout        = payout_manager
        self._report        = report_channel
        # v18.4 — optional second manager for ETH-side status/commands.
        # Additive only: existing /payout, /sweep etc. below still default
        # to self._payout (BSC) exactly as before. None on BSC-only
        # deployments (no PAYOUT_PRIVATE_KEY_ETH configured) — every call
        # site using this must handle that case rather than assume it's set.
        self._payout_eth    = payout_manager_eth
        # v17.22 (operator request — "can in telegram see Solana too") —
        # attached post-construction via attach_solana() once bot.py builds
        # the Solana leg (which happens after CommandHandlers today; see
        # that method's own docstring). None whenever SOLANA_SCAN_ENABLED
        # is off, exactly like self._payout_eth's own optional treatment.
        self._solana: Any | None = None
        self._dry_run       = dry_run
        self._min_profit    = min_profit
        self._scan_timeout  = scan_timeout
        self._oracle_url    = oracle_url.rstrip("/")
        # v17.21 — AI copilot state (/ai, /ai_idea → /approve).
        self._qwen          = qwen
        self._ai_pairs_path = ai_pairs_path
        self._ai_solana_routes_path = ai_solana_routes_path
        self._pending_ai_pair: dict | None = None
        # v18.36 — AI-action layer (/ai_do → /approve). Holds ONE validated
        # control action awaiting the operator's explicit /approve.
        self._pending_ai_action: dict | None = None

        # ── Mode state ─────────────────────────────────────────────────────
        self._ghost_mode = False   # when True: scan but don't execute
        self._mint_mode  = False   # when True: use aggressive (halved) profit threshold.
        # v17.20 FIX — this used to default True. effective_min_profit below
        # halves self._min_profit whenever this is True, so with the old
        # default every scan (manual /hunt AND the autonomous hunt loop,
        # since both call scan(min_profit=self.effective_min_profit)) ran
        # against $2.17 instead of the configured $4.35 the moment the bot
        # started — with no explicit action from the operator. Confirmed
        # exact match: 4.35 * 0.5 = 2.175 ≈ the $2.17 floor traced across
        # scanner.py v18.22-v18.25's diagnostics. Toggle stays available via
        # /mint_mode for anyone who wants the halved floor deliberately —
        # only the unrequested default changes here.

        # ── v17.14: scan decoupling ─────────────────────────────────────────
        # Shared lock so a manual /hunt and bot.py's scheduled auto-scan
        # loop never run concurrently. Exposed as a public property so
        # bot.py can do `async with handlers.scan_lock:` around its own
        # scanner.scan() call.
        self._scan_lock: asyncio.Lock = asyncio.Lock()
        self._bg_tasks: set[asyncio.Task] = set()

    # ── Public mode accessors (for bot.py / scheduled loop) ──────────────────

    @property
    def ghost_mode(self) -> bool:
        return self._ghost_mode

    @property
    def mint_mode(self) -> bool:
        return self._mint_mode

    @property
    def effective_min_profit(self) -> float:
        """Mint mode halves the minimum profit gate."""
        return self._min_profit * (0.5 if self._mint_mode else 1.0)

    @property
    def scan_lock(self) -> asyncio.Lock:
        """v17.14 — share this with bot.py's auto-scan loop (see module
        docstring) so manual and scheduled scans never overlap."""
        return self._scan_lock

    # ── Background task plumbing (v17.14) ─────────────────────────────────────

    def _spawn(self, coro) -> asyncio.Task:
        """
        Fire-and-forget a coroutine as a tracked background task. Keeps
        handle() fast-returning so a long-running /hunt scan never blocks
        whatever loop is awaiting handle() (typically Telegram polling)
        from seeing the next update.
        """
        task = asyncio.create_task(coro)
        self._bg_tasks.add(task)
        task.add_done_callback(self._bg_tasks.discard)
        return task

    def attach_solana(self, solana: Any) -> None:
        """bot.py calls this right after building the Solana leg (see that
        module's own Solana-leg section) — CommandHandlers is constructed
        earlier in startup, before self._solana exists there, so this is a
        post-construction attach rather than a constructor param. Lets
        /hunt surface Solana's own latest scan results alongside the ARB
        card instead of Telegram-side visibility being ARB-only."""
        self._solana = solana

    async def aclose(self) -> None:
        """
        Call from bot.py during graceful shutdown to let any in-flight
        background scan finish instead of being dropped mid-execution.
        """
        if self._bg_tasks:
            await asyncio.gather(*list(self._bg_tasks), return_exceptions=True)

    # ── Dispatcher ────────────────────────────────────────────────────────────

    async def handle(self, update: TelegramUpdate) -> None:
        # ★ SPINNER FIX ★ — ACK inline keyboard press immediately (≤5s required)
        if update.is_callback:
            try:
                await self._tg.answer_callback_query(update.callback_query_id)
            except Exception as exc:
                logger.warning("[CommandHandlers] answer_callback_query failed: %s", exc)

        chat_id = update.chat_id
        command = (update.command or "").strip().lower()

        # Plain reply-keyboard button presses arrive as text, not as a
        # parsed /command — map the button label to its command name.
        if not command and update.text:
            command = BUTTON_TO_COMMAND.get(update.text.strip(), "")

        if not command:
            if not update.is_callback:
                # v1.85 — DIRECT CHAT. Everything that was not a slash command
                # or an exact button label used to hit one line: "Unrecognized
                # input. Send /help or tap a button below." Thirty-odd buttons,
                # and no way to ask a question.
                #
                # The operator: "i need to open direct chat with bot in
                # telegram" — then, after trying: "i try /start hi how are you
                # / not answer".
                #
                # This is NOT the /ai path. bot.py's own v17.34 changelog
                # records why that was reined in: "operator forwarded several
                # 🤖-prefixed /ai replies inventing numbers with no basis in
                # this deployment". Every answer here is assembled from the
                # same live objects /doctor and /why read, so there is nothing
                # to hallucinate with. See modules/bot_chat.py.
                #
                # Gated behind the same owner check the commands use — this
                # reports custody state and blockers, and a stranger is not
                # entitled to either.
                answered = False
                if _OWNER_GATE_OPEN or chat_id in _OWNER_CHAT_IDS:
                    try:
                        from modules.bot_chat import get_chat

                        chat = get_chat()
                        chat.attach(self._solana)
                        reply = chat.reply(update.text or "")
                        if reply:
                            await self._safe_send(reply, chat_id)
                            answered = True
                    except Exception as exc:  # noqa: BLE001
                        logger.debug("[Chat] reply failed: %s", exc)
                if not answered:
                    await self._safe_send(
                        "I did not understand that one. I answer plain "
                        "questions like *why no trade*, *how are you*, *can "
                        "you sign*, *how fast are you*, *any signature*, "
                        "*how much profit*.\n\nOr /help for the command list.",
                        chat_id,
                    )
            return

        _stats["commands_total"] += 1
        args = update.args

        # ── OWNER gate (v17.40) — before everything, including the counter
        # of what an unauthorised chat "ran". Placed above the on-chain gate
        # because it strictly subsumes it: /chainstatus and /owner are also
        # commands, and reaching their narrower check means this one already
        # passed. See _load_owner_chat_ids() for why this exists and why its
        # fallback chain cannot lock the operator out.
        if (
            not _OWNER_GATE_OPEN
            and command not in _PUBLIC_COMMANDS
            and chat_id not in _OWNER_CHAT_IDS
        ):
            _stats["commands_total"] -= 1  # not a command this bot ran
            logger.warning(
                "[Auth] refused /%s from unauthorised chat_id=%s", command, chat_id,
            )
            await self._safe_send(
                "🔒 This bot is private.\n\n"
                "It is bound to its operator's chat and does not answer "
                "commands from anywhere else.",
                chat_id,
            )
            return

        # ── On-chain command gate (v17.16) — checked before dispatch, not
        # inside individual handlers, so a rejected attempt never touches
        # ContractManager or spends an RPC call. ──────────────────────────
        if command in _ON_CHAIN_COMMANDS and chat_id not in _ON_CHAIN_ALLOWED_CHAT_IDS:
            logger.warning(
                "Unauthorized on-chain command attempt: chat_id=%s command=%s",
                chat_id, command,
            )
            await self._safe_send(
                "Not authorized. This chat isn't on the on-chain command allowlist.",
                chat_id,
            )
            return

        try:
            if command == "start":
                # v17.41 — buttons only. /start used to follow the Command
                # Center with the entire HELP_TEXT: every command in the
                # bot, including a long Ethereum-era section this
                # deployment no longer uses, as one unscrollable wall on a
                # phone. Operator: "remove the list after start i need just
                # all tip". Every one of those commands has a tap button
                # directly above it, so the wall was never the way in — it
                # was a wall in front of it. /help still prints it for the
                # times you actually want the index.
                await self._tg.send_command_center(chat_id)
                # v1.85 — ANSWER THE QUESTION THEY ACTUALLY ASKED.
                #
                # The operator typed, verbatim: `/start hi how are you`, and
                # reported "not answer". They got the command centre — the
                # keyboard, correctly — and nothing at all addressing "hi how
                # are you", because args were parsed and then discarded.
                #
                # A person opening a chat with a question in it is asking the
                # question. Sending them a menu and dropping the question is
                # the machine answer to a human sentence.
                if args and (_OWNER_GATE_OPEN or chat_id in _OWNER_CHAT_IDS):
                    try:
                        from modules.bot_chat import get_chat

                        chat = get_chat()
                        chat.attach(self._solana)
                        reply = chat.reply(" ".join(args) if isinstance(args, list)
                                           else str(args))
                        if reply:
                            await self._safe_send(reply, chat_id)
                    except Exception as exc:  # noqa: BLE001
                        logger.debug("[Chat] /start reply failed: %s", exc)
                # v1.73 — TELL AN UNAUTHORISED CHAT HOW TO AUTHORISE ITSELF.
                #
                # Operator: "i need to open direct chat with bot in telegram."
                #
                # They could not, and the reason is subtle enough to be worth
                # writing down. The command gate authorises the FIRST source
                # that yields any id: TELEGRAM_OWNER_CHAT_IDS, then
                # TELEGRAM_ALLOWED_CHAT_IDS, then TELEGRAM_CHAT_ID. On this
                # deployment only TELEGRAM_CHAT_ID is set, and it holds the
                # CHANNEL the bot posts reports into — a different id from the
                # operator's own direct message with the bot.
                #
                # So the gate was working exactly as designed and still locked
                # the owner out of a private chat, and nothing said why: an
                # unauthorised chat just got the refusal, with no way to learn
                # the one number needed to fix it.
                #
                # /start is already public (it is in _PUBLIC_COMMANDS), so this
                # reveals nothing a stranger could not already learn — a chat
                # id is not a secret, it is the id of the chat you are already
                # in, and it is useless without shell access to the box.
                if _OWNER_CHAT_IDS and chat_id not in _OWNER_CHAT_IDS:
                    await self._safe_send(
                        "🔒 *this chat is not authorised yet*\n\n"
                        f"Your chat id is `{chat_id}`.\n\n"
                        "The bot currently answers only the channel it reports "
                        "into, which is a different chat from this one. To open "
                        "a direct chat, run this on the box and restart:\n\n"
                        f"`./scripts/setenv.sh TELEGRAM_OWNER_CHAT_IDS={chat_id}`\n\n"
                        "_To keep the channel working too, list both ids "
                        "comma-separated._",
                        chat_id,
                    )
            elif command == "help":
                await self._tg.send_command_center(chat_id)
                await self._safe_send(HELP_TEXT, chat_id)
            elif command == "stats":
                await self._cmd_stats(chat_id)
            elif command == "status":
                await self._cmd_status(chat_id)
            elif command == "hunt":
                await self._cmd_hunt(chat_id)
            elif command == "price":
                await self._cmd_price(chat_id)
            elif command == "arbitrage":
                await self._cmd_arbitrage(chat_id)
            elif command == "audit":
                await self._cmd_audit(chat_id)
            elif command == "flashloan":
                await self._cmd_flashloan(chat_id)
            elif command == "ghost_mode":
                await self._cmd_ghost_mode(chat_id)
            elif command == "mint_mode":
                await self._cmd_mint_mode(chat_id)
            elif command == "debug":
                await self._cmd_debug(chat_id)
            elif command == "circuit":
                await self._cmd_circuit(chat_id)
            elif command == "reset":
                await self._cmd_reset(chat_id)
            elif command == "force_reset":
                await self._cmd_force_reset(chat_id)
            elif command == "health":
                await self._cmd_health(chat_id)
            elif command == "doctor":
                await self._cmd_doctor(chat_id)
            elif command == "netdiag":
                await self._cmd_netdiag(chat_id)
            elif command == "speedtest":
                await self._cmd_speedtest(chat_id)
            elif command == "journal":
                await self._cmd_journal(chat_id)
            elif command == "ping":
                await self._cmd_ping(chat_id)
            elif command == "gas":
                await self._cmd_gas(chat_id)
            elif command == "testreport":
                await self._cmd_testreport(chat_id)
            elif command == "testlog":
                await self._cmd_testlog(chat_id)
            elif command == "chainstatus":
                await self._cmd_chainstatus(chat_id)
            elif command == "owner":
                await self._cmd_owner(chat_id)
            # v17.21 — payout/sweep/withdraw REMOVED from Telegram by
            # operator request (2026-07-12): anything that moves funds
            # now lives ONLY in the Hugging Face dashboard's Payout tab.
            elif command in ("payout", "payoutreset", "sweep",
                             "withdraw", "withdraweth"):
                await self._safe_send(
                    "💰 Payout, sweep and withdraw were removed from "
                    "Telegram (operator request). Use the Hugging Face "
                    "dashboard → Payout tab — funds can only be moved "
                    "from there now.",
                    chat_id,
                )
            # v17.21 — the AI copilot.
            elif command == "ai":
                await self._cmd_ai(chat_id, args)
            elif command == "ai_idea":
                await self._cmd_ai_idea(chat_id)
            elif command == "ai_idea_sol":
                await self._cmd_ai_idea_sol(chat_id)
            elif command == "ai_do":
                await self._cmd_ai_do(chat_id, args)
            elif command == "approve":
                await self._cmd_approve(chat_id)
            elif command == "reject":
                await self._cmd_reject(chat_id)
            # v17.36 (2026-07-28 operator review) — the Solana engine's new
            # subsystems, each with the one command that makes it
            # inspectable from the phone. Every one of these is READ-ONLY
            # except /resume, which only ever clears a pause.
            elif command == "routes":
                await self._cmd_routes(chat_id)
            elif command == "whatif":
                await self._cmd_whatif(chat_id, args)
            elif command == "jito":
                await self._cmd_jito(chat_id)
            elif command == "venues":
                await self._cmd_venues(chat_id)
            elif command == "resume":
                await self._cmd_resume(chat_id)
            elif command == "pumpfun":
                await self._cmd_pumpfun(chat_id)
            elif command == "decay":
                await self._cmd_decay(chat_id)
            elif command in ("nearmiss", "misses"):
                await self._cmd_nearmiss(chat_id)
            elif command in ("drawdown", "dd"):
                await self._cmd_drawdown(chat_id)
            elif command in ("capture", "catch"):
                await self._cmd_capture(chat_id)
            # v17.43 — the screens that answer the operator's actual
            # question. /doctor grades the configuration; these grade the
            # OUTCOME, which is a different question and the one that was
            # never being asked. See modules/why_report.py's own docstring.
            elif command == "why":
                await self._cmd_why(chat_id)
            elif command == "tune":
                await self._cmd_tune(chat_id, args)
            elif command in ("pipeline", "stages"):
                await self._cmd_pipeline(chat_id)
            elif command == "gates":
                await self._cmd_gates(chat_id)
            elif command in ("decisions", "why_not"):
                await self._cmd_decisions(chat_id)
            elif command == "bench":
                await self._cmd_bench(chat_id)
            elif command in ("lighthouse", "light", "stand"):
                await self._cmd_lighthouse(chat_id)
            elif command in ("observe", "observatory"):
                await self._cmd_observe(chat_id)
            elif command in ("tips", "tipmemory"):
                await self._cmd_tips(chat_id)
            elif command in ("sendpath", "speed", "400"):
                await self._cmd_sendpath(chat_id)
            elif command.startswith("tune_apply_"):
                await self._cmd_tune_apply(chat_id, command[len("tune_apply_"):])
            # v1.71 — the approval flow. `pc_ok_<token>` / `pc_no_<token>`
            # are the ONLY routes from a proposed change to the disk.
            elif command.startswith("pc_ok_"):
                await self._cmd_pending_decide(
                    chat_id, command[len("pc_ok_"):], approve=True)
            elif command.startswith("pc_no_"):
                await self._cmd_pending_decide(
                    chat_id, command[len("pc_no_"):], approve=False)
            elif command == "heal":
                await self._cmd_heal(chat_id)
            elif command == "verify":
                await self._cmd_verify(chat_id)
            elif command == "quote":
                await self._cmd_quote(chat_id, args)
            elif command == "next":
                await self._cmd_next(chat_id)
            elif command == "agent":
                await self._cmd_agent(chat_id)
            elif command in ("sig", "signatures"):
                await self._cmd_sig(chat_id)
            else:
                await self._safe_send(
                    f"Unknown command: /{command}\nSend /help for the full list.",
                    chat_id,
                )
        except Exception as exc:
            _stats["errors_total"] += 1
            logger.error("[CommandHandlers] /%s failed: %s", command, exc, exc_info=True)
            await self._safe_send(f"⚠️ /{command} failed: {str(exc)[:300]}", chat_id)

    # ── /status — overview snapshot ─────────────────────────────────────────────

    async def _cmd_status(self, chat_id: int) -> None:
        last = _recent_results[-1] if _recent_results else None
        last_signal = (last.get("signal") if last else None) or _stats["last_signal"]
        last_ts = _stats["last_scan_ts"]
        last_ts_str = (
            datetime.fromtimestamp(last_ts, tz=timezone.utc).strftime("%H:%M:%S UTC")
            if last_ts else "never"
        )
        # v17.41 — rebuilt to the shape the operator asked for: flat
        # `label: value` lines, Solana first, no section headers.
        #
        # What came out is what changed. The old version led with Ghost Mode
        # and Mint Mode (both OFF, both EVM-era), reported "Scans total" from
        # the retired ARB scanner rather than the Solana loop that is
        # actually running, and ended on "Scan lock held: False" — which
        # means nothing to anyone who has not read the source.
        #
        # A status screen's job is to answer "is it working and is it
        # earning". Everything below is on that question or is gone.
        sol = self._solana
        lines = ["📟 *status*\n"]

        dry = bool(getattr(getattr(self, "_bot", None), "config", None)
                   and getattr(self._bot.config, "dry_run", False))
        lines.append(f"mode: {'DRY RUN' if dry else 'LIVE'}")
        lines.append(f"uptime: {int(time.time() - _BOT_START_TIME) // 3600}h "
                     f"{(int(time.time() - _BOT_START_TIME) % 3600) // 60}m")

        if sol is not None:
            age = (f"{time.time() - sol.last_scan_ts:.0f}s ago"
                   if getattr(sol, "last_scan_ts", 0) else "never")
            lines.append(f"cycles: {sol.scan_count:,} · last {age}")
            lines.append(f"signals this run: {sol.signal_count:,}")
            lines.append(f"routes: {len(sol.routes)} configured")
            lines.append(f"floor: ${sol.min_profit:.2f} net after fees")
            paused = bool(getattr(sol, "paused", False))
            lines.append(f"scanning: {'PAUSED' if paused else 'active'}")

        # The two numbers that say whether it is EARNING, pulled from the
        # same source /capture uses so they can never disagree.
        try:
            from modules.capture_report import capture

            c = await asyncio.to_thread(capture)
            rate = c["capture_rate"]
            lines.append(
                f"capture: {'—' if rate is None else f'{rate:.1%}'} "
                f"({c['landed']}/{c['signalled']} signals)"
            )
            lines.append(f"seen ${c['gross_seen_usd']:,.2f} · "
                         f"taken ${c['net_taken_usd']:,.2f}")
        except Exception as exc:  # noqa: BLE001
            logger.debug("[/status] capture unavailable: %s", exc)

        try:
            from modules.trading_guard import get_guard

            gst = get_guard().status()
            if gst.get("state") != "closed":
                lines.append(f"⛔ guard: {str(gst.get('state')).upper()} — "
                             f"{(gst.get('trip_reason') or '')[:90]}")
        except Exception as exc:  # noqa: BLE001
            logger.debug("[/status] guard unavailable: %s", exc)

        lines.append(f"telegram: {'reachable' if self._tg.reachable else 'UNREACHABLE'}")
        if _stats["errors_total"]:
            lines.append(f"errors this session: {_stats['errors_total']}")
        lines.append(f"\n_last: {last_ts_str} — {last_signal}_")
        await self._tg.send_message("\n".join(lines), chat_id=chat_id, parse_mode="Markdown")

    # ── /hunt — non-blocking on-demand scan (v17.14 core fix) ───────────────────

    async def _cmd_hunt(self, chat_id: int) -> None:
        if not _arb_scan_enabled():
            # ARB retired (see _arb_scan_enabled) — report the Solana
            # leg's latest cycle instead of scanning a chain that can
            # only ever HOLD at current sizes.
            await self._safe_send(
                "🟣 ARB leg is retired (ARB_SCAN_ENABLED=false) — Solana-only mode.",
                chat_id,
            )
            if self._solana is not None:
                await self._send_solana_hunt_summary(chat_id)
            else:
                await self._safe_send(
                    "⏳ Solana leg not running either — nothing to scan.", chat_id,
                )
            return
        if self._scan_lock.locked():
            await self._safe_send(
                "⏳ A scan is already running (manual or scheduled) — try again shortly.",
                chat_id,
            )
            return

        await self._safe_send("🔍 Hunt started — scanning…", chat_id)
        self._spawn(self._run_hunt(chat_id))

    async def _run_hunt(self, chat_id: int) -> None:
        """Background task spawned by /hunt — see module docstring §v17.14."""
        async with self._scan_lock:
            try:
                try:
                    scan_coro = self._scanner.scan(min_profit=self.effective_min_profit)
                except TypeError:
                    # Scanner.scan() doesn't take min_profit as a kwarg on
                    # this build — fall back to a no-arg call.
                    scan_coro = self._scanner.scan()
                result = await asyncio.wait_for(scan_coro, timeout=self._scan_timeout)
            except asyncio.TimeoutError:
                _stats["errors_total"] += 1
                await self._safe_send(
                    f"⏱️ Scan timed out after {self._scan_timeout:.0f}s.", chat_id,
                )
                return
            except Exception as exc:
                _stats["errors_total"] += 1
                logger.error("[CommandHandlers] Hunt scan failed: %s", exc, exc_info=True)
                await self._safe_send(f"⚠️ Scan failed: {str(exc)[:300]}", chat_id)
                return

        # v17.26 — bookkeeping extracted into _record_scan_result() so the
        # autonomous hunt loop (run_autonomous_scan) records every cycle the
        # EXACT same way a manual /hunt does. Before this, only /hunt updated
        # these counters, so /status and the dashboard could only ever move
        # when a human clicked. See the §1.A note in _record_scan_result.
        result_dict = self._record_scan_result(result)

        send_result = result_dict
        if self._ghost_mode:
            send_result = dict(result_dict)
            send_result["reason"] = (
                (send_result.get("reason") or "") + " [ghost mode — not executed]"
            ).strip()

        await self._tg.send_opportunity(send_result, chat_id=chat_id)

        # v17.22 (operator request — "can in telegram see Solana too") —
        # see attach_solana()'s own docstring. Best-effort and additive:
        # Solana runs its own independent ~20s-interval loop
        # (solana_arb.py's run_loop), so this reads its latest
        # already-computed opportunities rather than triggering a second,
        # redundant scan cycle. A failure here must never affect the ARB
        # result already sent above.
        if self._solana is not None:
            await self._send_solana_hunt_summary(chat_id)

    async def _send_solana_hunt_summary(self, chat_id: int) -> None:
        """Solana section for /hunt — see _run_hunt's own comment for why
        this reads existing state instead of scanning.

        v17.37 (2026-07-29, operator wrote out the layout they wanted) —
        was three lines of `· hold SOL USDC->SOL->USDC: loan X -> Y (net Z,
        N bps)` and nothing else. Every number in it was true, and none of
        it answered "is it working, and if nothing is trading, what is
        stopping it?" — no posture, no momentum, no reason code, no view
        of the scanner's own budget, and only the top 3 of up to 14 pairs
        so a pruned route was indistinguishable from a broken one.

        Rendering lives in modules/hunt_view.py (see its docstring for what
        each part of the layout is for) so the dashboard can print the
        identical report without a second copy of the format strings.
        Falls back to the old summary if anything at all goes wrong — a
        formatting bug must never cost the operator their /hunt.
        """
        try:
            from modules.hunt_view import render

            await self._safe_send(render(self._solana.hunt_snapshot()), chat_id)
            return
        except Exception as exc:  # noqa: BLE001 — fall through to the plain view
            logger.warning(
                "[CommandHandlers] rich /hunt view failed (%s) — falling back "
                "to the plain summary", exc, exc_info=True,
            )
        try:
            opps = self._solana.last_opportunities
            last_ts = self._solana.last_scan_ts
            age = time.time() - last_ts if last_ts else None
            age_str = f"{age:.0f}s ago" if age is not None else "not scanned yet"
            if not opps:
                text = f"🟣 Solana — no opportunities from the last cycle ({age_str})."
            else:
                top = sorted(opps, key=lambda o: o.net_profit_usd, reverse=True)[:3]
                lines = [f"🟣 Solana (Jupiter round-trips) — last scan {age_str}:"]
                lines.extend(o.summary() for o in top)
                text = "\n".join(lines)
            await self._safe_send(text, chat_id)
        except Exception as exc:
            logger.debug("[CommandHandlers] Solana hunt summary failed: %s", exc)

    # ── Shared scan bookkeeping + autonomous hunt (v17.26) ──────────────────────

    def _record_scan_result(self, result) -> dict:
        """Shared post-scan bookkeeping for BOTH /hunt (_run_hunt) and the
        bot's own autonomous hunt loop (run_autonomous_scan). Updates the
        module stats and the recent-results ring buffer that feed /status,
        /audit, /arbitrage and the Gradio dashboard, then returns the
        normalized dict.

        FIX (Incident Report §1.A): `result` is a ScanResult dataclass
        returned by Scanner.scan(), not a dict — result.get("signal") raised
        AttributeError: 'ScanResult' object has no attribute 'get', crashing
        the pipeline right after a profitable opportunity was found (root
        cause of scans_buy staying at 0). Normalize to a plain dict once,
        here, so every downstream consumer works with dicts consistently.
        """
        _stats["scans_total"] += 1
        _stats["last_scan_ts"] = time.time()

        result_dict = result.to_dict() if hasattr(result, "to_dict") else dict(result)

        signal = result_dict.get("signal", "N/A")
        _stats["last_signal"] = signal
        if signal == "BUY":
            _stats["scans_buy"] += 1

        _recent_results.append(result_dict)
        if len(_recent_results) > _RECENT_CAP:
            del _recent_results[: len(_recent_results) - _RECENT_CAP]

        return result_dict

    async def run_autonomous_scan(self) -> Optional[str]:
        """v17.26 — ONE cycle of the bot's own 24/7 hunt loop (driven by
        bot.py's _autonomous_hunt_loop). This is the piece that was described
        all over the v17.14 scan-decoupling notes but never actually built:
        the scan_lock, effective_min_profit and every "scheduled auto-scan
        loop" comment already existed, yet nothing on the bot side ever
        called scan() on a timer — so the bot only scanned when a human sent
        /hunt or clicked the dashboard button. The instant the operator
        stopped interacting, hunting stopped (confirmed in production
        2026-07-10: last scan at 23:53:31, then 9.5h of nothing but Telegram
        polling). A real 24/7 hunter needs this.

        Silent by design: it updates the same stats/ring buffer /hunt does
        (so /status and the dashboard reflect autonomous scans) but does NOT
        send a per-cycle Telegram message — scanner.scan() already fires the
        signals that actually matter on its own (BUY execution + the
        "✅ LIVE TRADE EXECUTED" alert, near-miss alerts, and the hourly AI
        Market Pulse digest). A per-scan message here would just spam the
        chat every interval.

        Returns the signal string ('BUY'/'HOLD'/…) on a completed cycle, or
        None if a scan was already running or the cycle failed. NEVER raises
        — the caller is a forever-loop that must not die on one bad cycle.
        """
        if self._scan_lock.locked():
            # A manual /hunt (or the previous auto cycle) is still running —
            # skip this tick rather than queue behind it and pile up.
            return None

        async with self._scan_lock:
            try:
                try:
                    scan_coro = self._scanner.scan(min_profit=self.effective_min_profit)
                except TypeError:
                    # Scanner build without the min_profit kwarg — no-arg call.
                    scan_coro = self._scanner.scan()
                result = await asyncio.wait_for(scan_coro, timeout=self._scan_timeout)
            except asyncio.TimeoutError:
                _stats["errors_total"] += 1
                logger.warning(
                    "[CommandHandlers] Auto-hunt scan timed out after %.0fs.",
                    self._scan_timeout,
                )
                return None
            except Exception as exc:
                _stats["errors_total"] += 1
                logger.error(
                    "[CommandHandlers] Auto-hunt scan failed: %s", exc, exc_info=True,
                )
                return None

        result_dict = self._record_scan_result(result)
        return result_dict.get("signal", "N/A")

    # ── /price — live price snapshot ─────────────────────────────────────────────

    async def _cmd_price(self, chat_id: int) -> None:
        # FIX (Incident Report §1.C): get_prices() requires the `assets`
        # positional argument — calling it with no args raised
        # TypeError: PriceClient.get_prices() missing 1 required
        # positional argument: 'assets'. The defensive multi-method probe
        # is no longer needed now that the real signature is confirmed
        # (matches scanner.py and bot.py's usage: get_prices(list[str])).
        try:
            prices = await self._price.get_prices(["WETH", "BNB", "BTC"])
        except Exception as exc:
            logger.error("[CommandHandlers] /price fetch failed: %s", exc, exc_info=True)
            await self._safe_send(f"⚠️ Price fetch failed: {str(exc)[:300]}", chat_id)
            return

        # get_prices() returns dict[str, PriceEntry] (a dataclass, not a
        # dict) — send_price_report() does result.get(...) lookups on each
        # entry, which would hit the same AttributeError as §1.A. Normalize
        # defensively since price_client.py's exact PriceEntry fields
        # weren't available to verify directly.
        prices_dict: dict[str, dict[str, Any]] = {}
        for asset, entry in prices.items():
            if isinstance(entry, dict):
                prices_dict[asset] = entry
            else:
                prices_dict[asset] = {
                    "price":     getattr(entry, "price", 0.0),
                    "source":    getattr(entry, "source", "?"),
                    "is_live":   getattr(entry, "is_live", False),
                    "change24h": getattr(entry, "change24h", None),
                }

        await self._tg.send_price_report(prices_dict, chat_id=chat_id)

    # ── /arbitrage — last scan snapshot ──────────────────────────────────────────

    async def _cmd_arbitrage(self, chat_id: int) -> None:
        if not _recent_results:
            await self._safe_send("No scan results yet — run /hunt first.", chat_id)
            return
        await self._tg.send_opportunity(_recent_results[-1], chat_id=chat_id)

    # ── /audit — recent scan history (ring buffer) ───────────────────────────────

    async def _cmd_audit(self, chat_id: int) -> None:
        if not _recent_results:
            await self._safe_send("No scan history yet — run /hunt first.", chat_id)
            return
        lines = ["📜 *Recent Scan History*\n"]
        for r in _recent_results[-10:]:
            sig   = r.get("signal", "?")
            chain = r.get("chain", "?")
            icon  = "🚀" if sig == "BUY" else "😴"
            # ── Realized-Only gate (v2.0 reporting standard) ────────────────
            # A scanned spread on a non-BUY (no-trade) result is a
            # theoretical figure, not a realized profit. Showing it next to
            # an identical-looking number from an actual BUY is misleading
            # in an export/audit context. No trade = no profit display.
            if sig == "BUY":
                net = float(r.get("netAfterFee", r.get("net_profit", 0.0)))
                lines.append(f"{icon} `{chain}` net=`${net:,.2f}`")
            else:
                lines.append(f"{icon} `{chain}` no trade — signal=`{sig}`")
        await self._tg.send_message("\n".join(lines), chat_id=chat_id, parse_mode="Markdown")

    # ── /flashloan — cost-model assumptions ──────────────────────────────────────

    async def _cmd_flashloan(self, chat_id: int) -> None:
        text = (
            "⚡ *Flashloan Cost Model*\n\n"
            "Loan fee, gas cost, and net profit are computed per-scan by the "
            "Scanner and shown in the /hunt and /arbitrage results.\n"
            f"Current min-profit gate: `${self.effective_min_profit:.2f}` "
            f"({'mint-mode aggressive (halved)' if self._mint_mode else 'standard'})."
        )
        await self._tg.send_message(text, chat_id=chat_id, parse_mode="Markdown")

    # ── Mode toggles ──────────────────────────────────────────────────────────────

    async def _cmd_ghost_mode(self, chat_id: int) -> None:
        self._ghost_mode = not self._ghost_mode
        await self._tg.send_message(
            f"👻 Ghost Mode is now {'ON' if self._ghost_mode else 'OFF'}.",
            chat_id=chat_id, parse_mode=None,
        )

    async def _cmd_mint_mode(self, chat_id: int) -> None:
        self._mint_mode = not self._mint_mode
        await self._tg.send_message(
            f"🌙 Mint Mode is now {'ON' if self._mint_mode else 'OFF'} "
            f"(effective min profit: ${self.effective_min_profit:.2f}).",
            chat_id=chat_id, parse_mode=None,
        )

    # ── /ping ─────────────────────────────────────────────────────────────────────

    async def _cmd_ping(self, chat_id: int) -> None:
        await self._tg.send_message(
            f"🏓 Pong! Telegram: {'🟢' if self._tg.reachable else '🔴'}",
            chat_id=chat_id, parse_mode=None,
        )

    # ── /gas — last scan's gas estimate ──────────────────────────────────────────

    async def _cmd_gas(self, chat_id: int) -> None:
        if not _recent_results:
            await self._safe_send(
                "No scan data yet — run /hunt first to get a live gas estimate.",
                chat_id,
            )
            return
        last  = _recent_results[-1]
        gas   = float(last.get("gasCostUSD", last.get("gas_cost_usd", 0.0)))
        chain = last.get("chain", "?")
        await self._tg.send_message(
            f"⛽ Last estimated gas cost on `{chain}`: `${gas:,.4f}`",
            chat_id=chat_id, parse_mode="Markdown",
        )

    # ── /testreport — v18.42, operator report ("the channel for report not
    # working") ─────────────────────────────────────────────────────────────
    # Sends a real test message through ReportChannel and replies HERE, in
    # this same chat, with exactly what happened — the report channel
    # obviously can't show its own failure reason if it can't be posted to,
    # so that detail previously only ever reached the server log.

    async def _cmd_testreport(self, chat_id: int) -> None:
        if self._report is None:
            await self._safe_send(
                "⚠️ No report channel configured — neither TELEGRAM_REPORT_CHAT_ID "
                "nor REPORT_CHAT_ID is set. Set one to a chat/channel id the bot "
                "is an ADMIN in, then restart the Space.",
                chat_id,
            )
            return
        ok, detail = await self._report.send_test()
        if ok:
            await self._safe_send(
                "✅ Test message sent to the report channel successfully — "
                "go check it landed there.",
                chat_id,
            )
        else:
            await self._safe_send(f"❌ Report channel test FAILED: {detail[:600]}", chat_id)

    # ── /testlog — operator report ("help me let the channel telegram
    # working") ─────────────────────────────────────────────────────────────
    # TELEGRAM_LOG_CHANNEL_ID (PayoutManager → TelegramLogger) only ever
    # receives a real message on an actual payout/sweep event. An operator
    # with zero executed trades yet has never seen it fire once — no way to
    # tell "broken" from "never exercised" without a way to test it on
    # demand, same gap /testreport already closed for the report channel.

    async def _cmd_testlog(self, chat_id: int) -> None:
        if self._payout is None:
            await self._safe_send(
                "⚠️ No payout manager configured — nothing to test.",
                chat_id,
            )
            return
        ok, detail = await asyncio.to_thread(self._payout.test_telegram_log)
        if ok:
            await self._safe_send(
                "✅ Test message sent to the payout/log channel successfully — "
                "go check it landed there.",
                chat_id,
            )
        else:
            await self._safe_send(f"❌ Payout/log channel test FAILED: {detail[:600]}", chat_id)

    # ── On-chain commands (v17.16 merge, from command_handlers_5.py) ───────────
    #
    # ContractManager's web3 calls are synchronous/blocking (the default
    # HTTPProvider does real network I/O with no async support). Every call
    # below goes through asyncio.to_thread so a slow RPC node can't stall
    # the whole dispatcher the way an inline /hunt used to before v17.14 —
    # same lesson, same fix, applied to a new blocking call site.

    async def _cmd_chainstatus(self, chat_id: int) -> None:
        cm = get_contract_manager()
        try:
            owner, pool, balance = await asyncio.to_thread(
                lambda: (cm.get_owner(), cm.get_aave_pool(), cm.get_eth_balance())
            )
        except Exception as exc:
            await self._safe_send(f"Chain status failed: {str(exc)[:300]}", chat_id)
            return
        await self._tg.send_message(
            f"Owner: `{owner}`\nAave pool: `{pool}`\nContract ETH balance: `{balance}`",
            chat_id=chat_id, parse_mode="Markdown",
        )

    async def _cmd_owner(self, chat_id: int) -> None:
        cm = get_contract_manager()
        owner = await asyncio.to_thread(cm.get_owner)
        await self._tg.send_message(f"Owner: `{owner}`", chat_id=chat_id, parse_mode="Markdown")

    async def _cmd_debug(self, chat_id: int) -> None:
        cache_obj = (
            getattr(self._price, "cache", None)
            or getattr(self._price, "_cache", None)
            or getattr(self._price, "last_prices", None)
        )
        if isinstance(cache_obj, dict) and cache_obj:
            cache_lines = "\n".join(f"  {k}: {v}" for k, v in list(cache_obj.items())[:10])
        elif cache_obj:
            cache_lines = str(cache_obj)[:500]
        else:
            cache_lines = "(unavailable)"

        last = _recent_results[-1] if _recent_results else None
        text = (
            "Debug Snapshot\n\n"
            f"Stats: {json.dumps(_stats)}\n\n"
            f"Ghost mode: {self._ghost_mode} | Mint mode: {self._mint_mode}\n"
            f"Scan lock held: {self._scan_lock.locked()} | "
            f"Background tasks: {len(self._bg_tasks)}\n"
            f"Effective min profit: ${self.effective_min_profit:.2f}\n\n"
            f"Price cache:\n{cache_lines}\n\n"
            f"Last scan: {json.dumps(last, default=str) if last else 'none yet'}\n"
            f"Telegram reachable: {self._tg.reachable}"
        )
        await self._tg.send_message(text[:4000], chat_id=chat_id, parse_mode=None)

    async def _cmd_circuit(self, chat_id: int) -> None:
        text = (
            "Circuit Status\n\n"
            f"Telegram: {'🟢 closed (reachable)' if self._tg.reachable else '🔴 open (unreachable)'}\n"
            f"Ghost Mode: {'👻 ON' if self._ghost_mode else 'OFF'}\n"
            f"Mint Mode:  {'🌙 ON' if self._mint_mode else 'OFF'}\n"
            f"Errors this session: {_stats['errors_total']}"
        )
        await self._tg.send_message(text, chat_id=chat_id, parse_mode=None)

    async def _cmd_reset(self, chat_id: int) -> None:
        _stats["scans_total"]  = 0
        _stats["scans_buy"]    = 0
        _stats["errors_total"] = 0
        _stats["last_scan_ts"] = None
        _stats["last_signal"]  = "N/A"
        await self._tg.send_message("Scan stats reset.", chat_id=chat_id, parse_mode=None)

    async def _cmd_force_reset(self, chat_id: int) -> None:
        for key, val in (
            ("scans_total", 0), ("scans_buy", 0), ("commands_total", 0),
            ("errors_total", 0), ("last_scan_ts", None), ("last_signal", "N/A"),
        ):
            _stats[key] = val
        _recent_results.clear()
        self._price.invalidate()
        # Also reset modes to safe defaults
        self._ghost_mode = False
        self._mint_mode  = False   # v17.20 — was True; that's the halved-floor
        # mode, not the safe one (see __init__ note). Force-reset now restores
        # the actual safe default instead of silently re-enabling it.
        await self._tg.send_message(
            "Force reset complete — stats, scan history, price cache cleared.\n"
            "Ghost: OFF | Mint: OFF",
            chat_id=chat_id, parse_mode=None,
        )

    async def _cmd_health(self, chat_id: int) -> None:
        price_feed_ok = await self._price.health_check()
        text = (
            "Health Check\n\n"
            f"Price Feed: {'🟢' if price_feed_ok else '🔴'}\n"
            f"Telegram: {'🟢' if self._tg.reachable else '🔴'}\n"
            f"Ghost Mode: {'👻 ON' if self._ghost_mode else 'OFF'}\n"
            f"Mint Mode:  {'🌙 ON' if self._mint_mode else 'OFF'}\n"
            f"Uptime: {int(time.time() - _BOT_START_TIME)}s"
        )
        await self._tg.send_message(text, chat_id=chat_id, parse_mode=None)

    # ── Solana engine subsystems (v17.36) ────────────────────────────────────
    #
    # Six read-mostly commands, one per subsystem added in the 2026-07-28
    # review. Deliberately separate rather than folded into /doctor: /doctor
    # answers "can it trade right now", these answer "why is it behaving
    # the way it is", and cramming both into one message produced something
    # nobody reads on a phone.

    async def _cmd_routes(self, chat_id: int) -> None:
        """Per-route win rate mined from the trade journal — and which
        routes are currently pruned, with the numbers that pruned them."""
        from modules.route_scorer import get_scorer

        scorer = get_scorer()
        status = scorer.status()
        table = scorer.table()

        lines = ["📊 *routes* — per-route productivity\n"]
        if not status["enabled"]:
            lines.append(
                "⚠️ pruning is OFF (ROUTE_SCORE_PRUNING_ENABLED=false) — every "
                "route scans every cycle regardless of the numbers below.\n"
            )
        if not table:
            lines.append(
                f"No journal rows scored yet.\n`{status['journal_path']}`\n\n"
                f"Scoring needs {status['min_rows']} rows per route before it "
                f"will form an opinion. Leave the bot running and check back."
            )
            await self._safe_send("\n".join(lines), chat_id)
            return

        productive = [r for r in table if not r["pruned"]]
        pruned = [r for r in table if r["pruned"]]

        lines.append(
            f"{status['rows_read']:,} journal rows · {status['routes_scored']} "
            f"routes · {status['routes_pruned']} pruned\n"
        )
        for row in productive[:12]:
            mark = {"PRODUCTIVE": "🟢", "near-miss": "🟡", "cold": "⚪", "sampling": "🔵"}.get(
                row["verdict"], "⚪",
            )
            lines.append(
                f"{mark} `{row['route']}` — {row['signals']}/{row['rows']} signals "
                f"({row['signal_rate']:.1%}), mean ${row['mean_net_usd']:+.3f}, "
                f"best ${row['best_net_usd']:+.2f}, {row['mean_spread_bps']:+.1f} bps"
            )
        if pruned:
            lines.append(
                f"\n⏸ *pruned* — not scanned, re-probed every "
                f"{status['probe_every_cycles']} cycles:"
            )
            for row in pruned:
                lines.append(f"  • `{row['route']}` — {row['prune_reason']}")
            lines.append(
                "\n_Every pruned route is Jupiter call budget handed back to the "
                "routes above it. A pruned route reinstates itself the moment a "
                "re-probe produces a real signal._"
            )
        if status["pinned"]:
            lines.append(f"\n📌 pinned (never pruned): {', '.join(status['pinned'])}")
        await self._safe_send("\n".join(lines), chat_id)

    async def _cmd_whatif(self, chat_id: int, args: list[str]) -> None:
        """Replay a proposed config change against recorded history."""
        from modules.config_whatif import format_result, run

        try:
            result = run(list(args or []))
        except Exception as exc:  # noqa: BLE001 — a replay must never look like a bot crash
            logger.error("[CommandHandlers] /whatif failed: %s", exc, exc_info=True)
            await self._safe_send(f"🧪 whatif failed: {str(exc)[:250]}", chat_id)
            return
        await self._safe_send(format_result(result), chat_id)

    async def _cmd_jito(self, chat_id: int) -> None:
        """Dynamic tip auction state — the operator's own MEV report, live."""
        import os

        from modules.jito_tip_engine import LAMPORTS_PER_SOL, get_tip_engine

        status = get_tip_engine().status()
        enabled = os.getenv("SOLANA_JITO_ENABLED", "").strip().lower() in ("1", "true", "yes", "on")

        lines = ["⚡ *jito* — dynamic tip auction\n"]
        if not enabled:
            lines.append(
                "🔴 SOLANA_JITO_ENABLED is not true — bundles are OFF and every "
                "send goes out over ordinary RPC. Everything below is the "
                "engine's configured state, not live behaviour.\n"
            )
        lines.append(
            f"γ (aggressiveness): `{status['gamma']:.3f}` "
            f"{'(adaptive, ' if status['gamma_adaptive'] else '(fixed, '}"
            f"bounds {status['gamma_bounds'][0]:.2f}–{status['gamma_bounds'][1]:.3f})"
        )
        lines.append(
            f"  _γ is the share of net revenue bid to the validator. "
            f"Tip = min(max, max(floor, γ·(R−C)))._"
        )
        rate = status["landing_rate"]
        if rate is None:
            lines.append(
                f"\nLanding rate: not enough samples yet "
                f"({status['samples']} bundles) — γ holds until there are."
            )
        else:
            lines.append(
                f"\nLanding rate: `{rate:.0%}` over {status['samples']} bundles "
                f"(target {status['target_land_rate']:.0%})"
            )
            if rate < status["target_land_rate"]:
                lines.append("  ↗️ below target — γ is climbing (bidding harder)")
            elif rate > status["target_land_rate"]:
                lines.append("  ↘️ above target — γ is easing (keeping more edge)")
        floor = status["tip_floor_lamports"]
        if floor:
            lines.append(
                f"\nTip floor: `{floor:,}` lamports "
                f"({floor / LAMPORTS_PER_SOL:.6f} SOL) "
                f"— {status['tip_floor_source']}, {status['tip_floor_age_secs']}s old"
            )
        else:
            lines.append("\nTip floor: not fetched yet (or the feed is unreachable)")
        lines.append(
            f"Ceiling: `{status['tip_max_lamports']:,}` lamports · "
            f"min keep: `{status['min_keep_lamports']:,}`"
        )
        lines.append(
            f"\nBundles: {status['bundles_landed']}/{status['bundles_submitted']} landed · "
            f"{status['priced_out']} priced out · "
            f"{status['tip_sol_paid']:.6f} SOL paid in tips"
        )
        engines = status["block_engines"]
        lines.append(f"\nBlock engines ({len(engines)}, submitted in parallel):")
        for url in engines:
            host = url.replace("https://", "").split(".")[0]
            lines.append(f"  • `{host}`")
        lines.append(
            "\n_Redundant submission is free — the auction dedupes identical "
            "bundles, so N relays is one entry at one tip._"
        )
        await self._safe_send("\n".join(lines), chat_id)

    async def _cmd_venues(self, chat_id: int) -> None:
        """Live probe of the direct-DEX fallback adapters."""
        import httpx

        from modules.dex_venues import get_router

        router = get_router()
        status = router.status()

        lines = ["🔀 *venues* — direct DEX quote fallback\n"]
        if not status["enabled"]:
            lines.append(
                "🔴 DEX_FALLBACK_ENABLED=false — Jupiter is the only price "
                "source. When it rate-limits, the whole Solana leg goes dark.\n"
            )
        else:
            lines.append(
                "🟢 fallback armed — used ONLY when Jupiter is cooling or its "
                "hourly budget is spent, never as a race against it.\n"
            )

        await self._safe_send("\n".join(lines) + "\nProbing venues…", chat_id)
        try:
            async with httpx.AsyncClient() as client:
                probes = await router.probe(client)
        except Exception as exc:  # noqa: BLE001
            await self._safe_send(f"⚠️ probe failed: {str(exc)[:200]}", chat_id)
            return

        out = ["🔀 *venues* — live probe (1 SOL → USDC)\n"]
        for row in probes:
            mark = "🟢" if row.get("ok") else "🔴"
            state = "on" if row["enabled"] else "off"
            out.append(
                f"{mark} `{row['venue']}` ({state}, {row.get('latency_ms', '?')}ms)\n"
                f"    {row.get('detail', '')[:200]}"
            )
        # v17.52 — probe the path that actually prices, not only the dead
        # fallbacks. Orca and Meteora have no public HTTP quote API left
        # (see dex_venues.py — one 404s, the other answers 200 with a
        # 16-month-old cached Cloudflare error). But both are quoted every
        # cycle through Jupiter's `dexes=` filter, which is what
        # scan_crossvenue uses. Showing only the dead adapters told the
        # operator the bot was blind to venues it prices constantly.
        try:
            jup = await self._probe_jupiter_venues()
        except Exception as exc:  # noqa: BLE001
            jup = []
            logger.debug("[Venues] jupiter dexes probe failed: %s", exc)
        if jup:
            out.append("\n*Via Jupiter `dexes=` — the path actually used:*")
            for name, ok, detail in jup:
                out.append(f"{'🟢' if ok else '🔴'} `{name}` {detail}")

        failing = [r["venue"] for r in probes if not r.get("ok")]
        if failing:
            # v17.51 — SAY WHAT A RED LINE HERE ACTUALLY COSTS.
            #
            # The operator read these 404s as "the bot cannot see Orca or
            # Meteora" and asked twice to have them turned on. They are
            # already on, through the path the bot actually prices with:
            # scan_crossvenue asks Jupiter with `dexes=Whirlpool`,
            # `dexes=Orca V2`, `dexes=Meteora DLMM`, and those quotes come
            # back — the operator's own run named "Orca V2" and "Whirlpool"
            # as legs.
            #
            # These direct adapters are an OFFLINE FALLBACK for a Jupiter
            # outage and nothing else. Reporting them in red with no context
            # made a spare tyre look like a flat one.
            # v1.72 — one calm line, not a warning banner. Orca and Meteora
            # are no longer registered at all (see dex_venues._build_venues),
            # so anything red here is a venue that was deliberately enabled
            # and is genuinely failing — which IS worth a warning.
            out.append(
                f"\n⚠️ _{', '.join(failing)}: enabled but not answering._"
            )
            out.append(
                "\n*The bot is not blind to them* — see the Jupiter block "
                "above. `ORCA_API_BASE` / `METEORA_API_BASE` still override "
                "the host if either ever publishes one again."
            )
        passing = [r["venue"] for r in probes if r.get("ok") and not r["enabled"]]
        if passing:
            out.append(
                f"\n✅ probing clean but disabled: {', '.join(passing)} — enable "
                f"with `DEX_VENUE_{passing[0].upper()}=true`"
            )
        await self._safe_send("\n".join(out), chat_id)

    async def _cmd_resume(self, chat_id: int) -> None:
        """Clear the trading circuit breaker."""
        from modules.trading_guard import get_guard

        guard = get_guard()
        before = guard.status()
        if before["state"] == "closed":
            await self._safe_send(
                "✅ trading circuit breaker is already CLOSED — nothing to "
                "resume. Trading is allowed; if nothing is executing, "
                "run /doctor.",
                chat_id,
            )
            return
        detail = guard.reset(why=f"/resume from chat {chat_id}")
        logger.warning("[CommandHandlers] /resume — %s", detail)
        await self._safe_send(
            f"▶️ *Trading resumed.*\n\n{detail}\n\n"
            f"It tripped on: {before['trip_reason'][:300]}\n\n"
            f"_If the underlying cause is still there it will trip again — "
            f"the cooldown restarts from its base, not from the doubled "
            f"value it had reached._",
            chat_id,
        )

    async def _cmd_pumpfun(self, chat_id: int) -> None:
        """pump.fun verification API status + live credential probe."""
        import httpx

        from modules.pumpfun_client import get_client, is_configured

        client_obj = get_client()
        status = client_obj.status()
        lines = ["🅿️ *pump.fun* — verification API\n"]
        if not is_configured():
            lines.append(
                "🔴 `PUMPFUN_API_KEY` is not set.\n\n"
                "Set it in .env (or Space Variables) and restart. The key is "
                "origin-restricted, so also check `PUMPFUN_ORIGIN` matches one "
                "of the allowed origins on the key "
                "(elghaly.dev / arb.elghaly.dev / bot.elghaly.dev)."
            )
            await self._safe_send("\n".join(lines), chat_id)
            return
        lines.append(
            f"key `{status['key_fingerprint']}` · origin `{status['origin']}`\n"
            f"{status['calls']} call(s), {status['errors']} error(s), "
            f"{status['cached_mints']} mint(s) cached"
        )
        await self._safe_send("\n".join(lines) + "\n\nProbing…", chat_id)
        try:
            async with httpx.AsyncClient() as http:
                probe = await client_obj.probe(http)
        except Exception as exc:  # noqa: BLE001
            await self._safe_send(f"⚠️ probe failed: {str(exc)[:200]}", chat_id)
            return
        mark = "🟢" if probe.get("reachable") else "🔴"
        await self._safe_send(
            f"{mark} *pump.fun probe* ({probe.get('latency_ms', '?')}ms)\n\n"
            f"{probe.get('detail', '')[:400]}",
            chat_id,
        )

    async def _cmd_decay(self, chat_id: int) -> None:
        """How much edge dies between detection and send — measured, not
        guessed. See modules/decay_tracker.py."""
        from modules.decay_tracker import get_tracker

        st = get_tracker().status()
        lines = ["⏱️ *decay* — edge lost between detection and send\n"]
        if not st["recording"]:
            lines.append("🔴 not recording (DECAY_RECORD_ENABLED=false)")
            await self._safe_send("\n".join(lines), chat_id)
            return
        if st["samples"] < st["min_samples"]:
            lines.append(
                f"{st['samples']}/{st['min_samples']} real execution attempts "
                f"recorded so far.\n\n"
                f"This measures the one thing that has never been measured here: "
                f"what the scanner saw, what the pre-send re-quote saw, and how "
                f"long sat between them. Every 'signal fired but nothing sent' in "
                f"this bot's history is that gap. It needs "
                f"{st['min_samples'] - st['samples']} more attempt(s)."
            )
            await self._safe_send("\n".join(lines), chat_id)
            return

        # v17.42 — the RECENT slice first. The 200-sample window hydrates
        # from disk, so a pipeline change is invisible in the p90 for hours
        # and can even look like a regression (14.64s -> 15.63s right after
        # the fix that removed ~11s of queueing — same old samples, plus
        # one). Leading with the last N answers "is it better now", which
        # is the question actually being asked.
        rec, earlier = st.get("recent") or {}, st.get("earlier") or {}
        if rec.get("median_secs") is not None:
            lines.append(
                f"*last {rec['samples']} attempts: {rec['median_secs']:.2f}s* "
                f"median pipeline"
            )
            if earlier.get("median_secs"):
                delta = st.get("pipeline_trend_secs")
                arrow = "🟢 faster" if (delta or 0) < -0.5 else (
                    "🔴 slower" if (delta or 0) > 0.5 else "flat")
                lines.append(
                    f"  earlier {earlier['samples']} attempts: "
                    f"{earlier['median_secs']:.2f}s → {arrow} "
                    f"by {abs(delta or 0):.2f}s"
                )
            else:
                lines.append("  _no earlier samples to compare against yet_")
            lines.append("")

        lines.append(
            f"decay rate: *${st['decay_usd_per_sec']:.4f}/s* "
            f"(median over {st['samples']} attempts)"
        )
        lines.append(
            f"pipeline, WHOLE window: {st['pipeline_median_secs']:.2f}s median, "
            f"*{st['pipeline_p90_secs']:.2f}s p90*"
        )
        share = st.get("grew_share")
        if share is not None:
            lines.append(
                f"edge GREW on *{st['grew_count']}* of {st['samples']} attempts "
                f"({share * 100:.0f}%)"
            )
            if share >= 0.35:
                lines.append(
                    "_Adversarial decay cannot run that close to a coin flip. "
                    "At this share the median is mostly quote-to-quote noise, "
                    "not competitors taking the edge — which means more latency "
                    "work will not move it. Widen the edge, don't shorten the "
                    "wire._"
                )
            else:
                lines.append(
                    "_Skewed toward shrinking, which is what real competition "
                    "looks like. Latency work pays here._"
                )
        lines.append(
            f"\n→ the gate charges *${st['implied_adder_usd']:.3f}* of headroom above "
            f"the floor, from {st['gate_window']} recent attempt(s)."
            + ("  ⚠️ *CAPPED* — the raw figure is higher; read it as "
               "\"more than this\", not as a measurement."
               if st.get("gate_capped") else "")
        )
        lines.append(
            "_This is the p90 of dollars ACTUALLY lost between the two quotes, "
            "at the same size. It used to be rate × pipeline-p90, which broke "
            "once the pipeline got fast: the rate is dollars ÷ seconds, so the "
            "same fixed quote noise reads as a bigger rate the shorter the "
            "pipeline gets. Measuring the dollars directly cannot drift that way._"
        )
        if st["routes"]:
            lines.append("\nper route:")
            for r in st["routes"]:
                lines.append(
                    f"  `{r['route']}` — {r['attempts']} attempt(s), "
                    f"median decay ${r['median_decay_usd']:+.3f} over "
                    f"{r['median_elapsed_ms']:,.0f}ms, {r['survived']} survived"
                )
        lines.append(
            f"\ngate: {'ON — floor raised by the measurement above' if st['gate_enabled'] else 'OFF (recording only) — set DECAY_GATE_ENABLED=true to enforce'}"
        )
        await self._safe_send("\n".join(lines), chat_id)

    async def _cmd_nearmiss(self, chat_id: int) -> None:
        """Why signals are not becoming trades (v17.39).

        The one question a P&L cannot answer: are we losing races, or not
        finding opportunities? Both read as $0.00. See
        modules/near_miss.py for the taxonomy.
        """
        from modules.near_miss import get_log

        s = get_log().summary()
        lines = ["🥀 *near misses* — signals that never became receipts\n"]

        if not s["total"]:
            lines.append(s["verdict"])
            lines.append(
                "\n_Nothing to fix here yet. This log fills only when a signal "
                "clears the floor and then fails to land — until then the "
                "constraint is upstream, in finding an edge at all._"
            )
            await self._safe_send("\n".join(lines), chat_id)
            return

        fam = s["by_family"]
        total = s["total"]
        lines.append(
            f"*{total:,}* recorded (window {s['window']:,}), "
            f"{s['oldest_ts'][:16].replace('T', ' ')} → "
            f"{s['newest_ts'][:16].replace('T', ' ')}\n"
        )
        for name, label, emoji in (
            ("lost", "lost — beaten or decayed", "🔴"),
            ("blocked", "blocked — our own gating", "🟠"),
            ("broke", "broke — malfunction", "⚪"),
        ):
            count = fam.get(name, 0)
            if count:
                lines.append(f"{emoji} {label}: *{count:,}* ({count / total:.0%})")

        lines.append("\nby outcome:")
        for code, count in list(s["by_outcome"].items())[:8]:
            lines.append(f"  `{code}` — {count:,}")

        if s["value_at_risk_usd"] > 0:
            lines.append(
                f"\nscanner valued these at *${s['value_at_risk_usd']:,.2f}* in total "
                f"at detection. That is not a loss — most never sent — it bounds "
                f"the size of the prize being missed."
            )

        lines.append(f"\n{s['verdict']}")
        await self._safe_send("\n".join(lines), chat_id)

    async def _cmd_stats(self, chat_id: int) -> None:
        """One screen for "how is it going" (v17.41).

        Operator, repeatedly: "the stats is the most important". The numbers
        all existed already, spread across /routes, /nearmiss, /decay,
        /capture and /journal — five commands on a phone to assemble one
        answer. See modules/stats_report.py.
        """
        from modules.stats_report import render_stats

        text = await asyncio.to_thread(render_stats, self._solana)
        await self._safe_send(text, chat_id)

    async def _cmd_capture(self, chat_id: int) -> None:
        """The funnel: signalled -> attempted -> landed, and where it leaks.

        Operator: "there's signal and there's profit why he not catch — we
        need to know the reason ... how many % he can catch." No existing
        command answered that: /routes counts signals, /nearmiss counts
        failures, /decay counts seconds, and none of them divides one by
        another. See modules/capture_report.py.
        """
        from modules.capture_report import render_capture

        text = await asyncio.to_thread(render_capture)
        await self._safe_send(text, chat_id)

    # ── v17.43 — outcome screens, not configuration screens ──────────────
    async def _cmd_why(self, chat_id: int) -> None:
        """One screen: why nothing traded, ranked by where in the funnel it
        bites, ending in a numbered order to fix it in.

        Operator, repeatedly: "why no trade and what happen to the routes",
        "i lose the track before start", "the doctor say use this order to
        fix". /doctor answers "is anything misconfigured", which on
        2026-07-31 was honestly "no" while the scanner sat dark and 17 of 17
        attempts had landed nothing. A configuration check structurally
        cannot notice that nothing is happening. See modules/why_report.py.
        """
        from modules.why_report import report_text

        text = await asyncio.to_thread(report_text, self._solana)
        await self._safe_send(text, chat_id)

    async def _cmd_tune(self, chat_id: int, args: Any = None) -> None:
        """Ranked recommendations, each replayed against real history where
        it can be, each one tap from being applied.

        `/tune undo` restores the backup scripts/setenv.sh took.

        v17.44 — `args` is a LIST of tokens (TelegramClient's
        `update.args: list[str]`), not a string. The first version called
        .strip() on it and every /tune crashed with "'list' object has no
        attribute 'strip'". Accepts both shapes now, because /whatif takes
        the list form and a future caller passing a string should not be a
        second outage.
        """
        from modules import tune_report

        if isinstance(args, (list, tuple)):
            text = " ".join(str(a) for a in args)
        else:
            text = str(args or "")
        if text.strip().lower().startswith("all"):
            ok, message = await asyncio.to_thread(tune_report.apply_all, self._solana)
            await self._safe_send(message, chat_id)
            return
        if text.strip().lower().startswith("undo"):
            ok, message = await asyncio.to_thread(tune_report.undo)
            await self._safe_send(message, chat_id)
            return

        recs = await asyncio.to_thread(tune_report.build, self._solana)
        text = await asyncio.to_thread(tune_report.render, recs)
        # One Apply button per applyable recommendation, numbered to match
        # the list. callback_data carries the recommendation's id, not its
        # position — a list that reorders between render and tap must not
        # apply the wrong change.
        buttons = [
            {"text": f"Apply {i}", "callback_data": f"tune_apply_{rec.id}"}
            for i, rec in enumerate(recs, 1) if rec.applyable
        ][:6]
        markup = None
        if buttons:
            rows = [buttons[i:i + 3] for i in range(0, len(buttons), 3)]
            # v17.47 — operator: "allow apply from telegram for all". The
            # one-at-a-time rule exists so a change can be MEASURED, and
            # that is right — but it is a luxury while the bot has not
            # traded at all, and five taps between restarts on a phone is
            # its own failure. Both are offered; apply_all() says out loud
            # what it gave up.
            if len(buttons) > 1:
                rows.append([{"text": f"⚡ Apply ALL {len(buttons)}",
                              "callback_data": "tune_apply_ALL"}])
            markup = {"inline_keyboard": rows}
        try:
            await self._tg.send_message(
                text, chat_id=chat_id, parse_mode=None, reply_markup=markup,
            )
        except Exception as exc:  # noqa: BLE001
            logger.error("[CommandHandlers] /tune send failed: %s", exc)
            await self._safe_send(text, chat_id)

    async def _cmd_tune_apply(self, chat_id: int, rec_id: str) -> None:
        """Turn an Apply tap into a PROPOSAL, not a write.

        v1.71 — this used to write .env on a single tap, with an "Apply ALL"
        button beside it. The operator asked for the opposite: "should not
        follow all what he said, so the system should ask me y/n."

        So the tap now produces a card showing `was -> now` and asking for a
        second, explicit yes. The write itself still goes through
        setenv.sh (backup, verify, undo) exactly as before — what changed is
        that a decision is now made with the current value visible, rather
        than after the fact.
        """
        from modules import tune_report
        from modules.pending_change import get_store

        if rec_id == "ALL":
            # Deliberately NOT a bulk write any more. "Apply ALL" was one tap
            # that could change six settings, which is the exact opposite of
            # being asked about each one.
            recs = await asyncio.to_thread(tune_report.build, self._solana)
            applyable = [r for r in recs if r.applyable]
            if not applyable:
                await self._safe_send("Nothing is one-key applyable right now.",
                                      chat_id)
                return
            await self._safe_send(
                f"Proposing {len(applyable)} change(s) one at a time — each "
                f"needs its own yes.", chat_id,
            )
            for rec in applyable[:6]:
                await self._propose_change(
                    chat_id, rec.key, rec.value, rec.title, "tune",
                    detail=rec.caveat or "",
                    needs_restart="restart" in (rec.caveat or "").lower(),
                )
            return

        recs = await asyncio.to_thread(tune_report.build, self._solana)
        rec = next((r for r in recs if r.id == rec_id), None)
        if rec is None:
            await self._safe_send(
                f"`{rec_id}` is no longer recommended — the state behind it "
                f"changed. Run /tune again for the current list.", chat_id,
            )
            return
        if not rec.applyable:
            await self._safe_send(
                f"*{rec.title}* is not a one-key change.\n\n{rec.caveat or ''}",
                chat_id,
            )
            return
        await self._propose_change(
            chat_id, rec.key, rec.value, rec.title, "tune",
            detail=rec.caveat or "",
            needs_restart="restart" in (rec.caveat or "").lower(),
        )

    # ── the approval flow ───────────────────────────────────────────────────
    async def _propose_change(
        self, chat_id: int, key: str, value: Any, title: str, source: str,
        *, detail: str = "", needs_restart: bool = False,
    ) -> None:
        """Send a was -> now card with Apply / Reject. Writes nothing."""
        from modules.pending_change import get_store

        change = get_store().propose(
            key, value, title, source, detail=detail,
            needs_restart=needs_restart, chat_id=chat_id,
        )
        if str(change.was).strip() == str(value).strip():
            await self._safe_send(
                f"*{key}* is already `{value}` — nothing to change.", chat_id,
            )
            get_store().reject(change.token)
            return
        await self._tg.send_message(
            change.card(), chat_id=chat_id, parse_mode=None,
            reply_markup=change.keyboard(),
        )

    async def _cmd_pending_decide(self, chat_id: int, token: str,
                                  approve: bool) -> None:
        """The second tap. This is the only path from a proposal to disk."""
        from modules.pending_change import get_store

        store = get_store()
        if not approve:
            change = store.reject(token)
            await self._safe_send(
                f"✖️ rejected — *{change.key}* left at `{change.was or 'unset'}`"
                if change else
                "✖️ rejected (that proposal had already expired).",
                chat_id,
            )
            return

        ok, message = await asyncio.to_thread(store.apply, token)
        await self._safe_send(message, chat_id)
        if not ok:
            return

        change_needs_restart = "RESTART" in message
        if change_needs_restart:
            # A restart is a SEPARATE decision, with its own tap. Bundling it
            # into the value change would mean one yes bought two actions.
            restart = store.propose(
                "", "", "restart the bot so that takes effect", "tune",
                detail="The setting is written. The running process is still "
                       "using the old value until it restarts.",
                chat_id=chat_id, action=self._restart_action(),
            )
            await self._tg.send_message(
                restart.card(), chat_id=chat_id, parse_mode=None,
                reply_markup=restart.keyboard(),
            )

    def _restart_action(self):
        """Returns a callable that restarts the service. Approval-gated.

        Python has never restarted this service — only `gat set` did, via
        systemctl. Doing it from Telegram is what the operator asked for
        ("make the system restart from telegram to apply the change"), and it
        is safe only because it cannot be reached without an explicit tap on
        a card that says exactly what it will do.
        """
        def _do() -> tuple[bool, str]:
            import shutil
            import subprocess

            unit = os.getenv("GA_SYSTEMD_UNIT", "garden-angel-terminal")
            if not shutil.which("systemctl"):
                return False, (
                    "systemctl is not available here, so I cannot restart the "
                    f"service. Run it yourself:\n`sudo systemctl restart {unit}`"
                )
            try:
                result = subprocess.run(
                    ["sudo", "-n", "systemctl", "restart", unit],
                    capture_output=True, text=True, timeout=30,
                )
            except (OSError, subprocess.SubprocessError) as exc:
                return False, f"restart failed: {str(exc)[:200]}"
            if result.returncode != 0:
                return False, (
                    f"restart failed: {(result.stderr or result.stdout)[:250]}\n\n"
                    f"Run it yourself:\n`sudo systemctl restart {unit}`"
                )
            return True, (
                f"♻️ restarting `{unit}` — the new value is live once it comes "
                f"back. This message may be the last thing sent by the old "
                f"process."
            )
        return _do

    async def _cmd_sig(self, chat_id: int) -> None:
        """Every signature this bot has put on chain, as solscan links.

        The operator: "i need to see his signature on chain".

        There was no way to. The signature is written into the trade
        journal's free-text `detail` column and nothing ever read it back
        out — it appeared once, in one Telegram message, at the moment of
        the send. If you were asleep, it was gone.

        This deliberately points at scripts/verify_signatures.py at the end.
        A list of signatures I printed is a list of things I BELIEVE
        happened; that script asks the chain. On a bot with 20 attempts and
        0 confirmed landings, the difference between those two is the entire
        question.
        """
        from modules.bot_chat import executed_signatures

        rows = executed_signatures(10)
        if not rows:
            await self._safe_send(
                "🔗 *signatures* — none yet.\n\n"
                "No journal row is marked executed with a signature in it. "
                "That matches /doctor: attempts recorded, none confirmed on "
                "chain.\n\n"
                "/gates shows what is holding them, /why ranks the cause.",
                chat_id)
            return
        lines = ["🔗 *signatures on chain*", ""]
        for sig, when, detail in rows:
            lines.append(f"• [`{sig[:24]}…`](https://solscan.io/tx/{sig})")
            lines.append(f"     _{when} · {detail[:100]}_")
        lines += [
            "",
            "_These are what I recorded. To check what the CHAIN says — "
            "which is the only thing that counts:_",
            "`venv/bin/python scripts/verify_signatures.py`",
        ]
        await self._safe_send("\n".join(lines), chat_id)

    async def _cmd_agent(self, chat_id: int) -> None:
        """Is the watcher watching, and what is it watching?

        The agent's whole value is that it stays quiet. That makes it
        indistinguishable, from the outside, from an agent that died three
        days ago — which is the same silence-reads-as-health failure it was
        built to fix. So there has to be a way to ask.
        """
        try:
            from modules.agent_loop import INTERVAL_SECS, REPEAT_SECS, get_agent

            agent = get_agent()
            if agent is None:
                await self._safe_send(
                    "🤖 *agent* — not started\n\nThe Solana leg has to be "
                    "running for it to have anything to watch. /doctor.",
                    chat_id)
                return
            st = agent.status()
            lines = [
                "🤖 *agent* — the bot watching itself", "",
                f"{'🟢 running' if st['running'] else '🔴 STOPPED'} · "
                f"{st['ticks']} check(s) · {st['announcements']} message(s) sent",
                f"checks every {INTERVAL_SECS / 60:.0f} min, speaks only when "
                f"something CHANGES, re-raises an unchanged fault every "
                f"{REPEAT_SECS / 3600:.0f}h",
                "",
            ]
            if st["watching"]:
                lines.append("*currently blocking:*")
                lines += [f"  • {h}" for h in st["watching"][:5]]
                if st["unchanged_for_secs"]:
                    lines.append(
                        f"\n_unchanged for "
                        f"{st['unchanged_for_secs'] / 60:.0f} min — that is why "
                        f"it has gone quiet, not because it stopped looking._"
                    )
            else:
                lines.append("*nothing is blocking right now.*")
            lines += [
                "", "_It never writes a setting. Every fix arrives as the "
                "same y/n card /heal produces._",
            ]
            await self._safe_send("\n".join(lines), chat_id)
        except Exception as exc:  # noqa: BLE001
            await self._safe_send(f"🤖 agent unavailable: {str(exc)[:150]}", chat_id)

    def primary_chat_id(self) -> Optional[int]:
        """Where an UNPROMPTED message should go, or None.

        Every other send in this class replies to a chat_id that arrived
        with a command. The agent has no such chat: it speaks first. It
        needs the operator's own chat, and it must never guess — a proposal
        card offering to change trading settings, delivered to the wrong
        chat, is a stranger being handed the controls.

        So: the same authorised set the command gate itself uses, and the
        lowest id from it for stability (the set is unordered, and an agent
        that addresses a different chat on each restart is one the operator
        cannot follow). None when nothing is configured, which correctly
        makes the agent report-only rather than making it improvise.
        """
        if not _OWNER_CHAT_IDS:
            return None
        return min(_OWNER_CHAT_IDS)

    async def _cmd_heal(self, chat_id: int) -> None:
        """Find everything fixable and offer each fix as one tap.

        v1.76 — operator, repeatedly: "make the system fix himself by
        running doctor tune decay", "i run doctor he have problem then fix
        its mean the system heal him self under my accept".

        Everything needed for this already existed and was never joined.
        /doctor finds problems and prints commands to run by hand. /tune
        ranks changes and, until this branch, applied them on a single
        untracked tap. /gates and /decay each hold an opinion and no way to
        act on it. On a phone that is five screens, a lot of scrolling, and
        a shell prompt the operator does not have.

        So: one button. It collects every ACTIONABLE finding — meaning one
        with a concrete key and value, not a paragraph of advice — and sends
        each as a was -> now card with Apply / Reject.

        It never applies anything itself. That is the whole point of the
        word "under my accept": the bot is allowed to find, rank and
        propose, and the operator is the only thing that writes. A bot that
        edits its own configuration unattended is a bot whose behaviour
        cannot be reasoned about afterwards, which is the argument
        tune_report has made since it was written and this keeps.
        """
        await self._safe_send("🩹 checking everything fixable…", chat_id)

        proposals: list[tuple[int, str, str, Any, str, bool]] = []
        # (priority, key, value, _, title, needs_restart)

        # 1 — /tune's ranked, applyable recommendations.
        try:
            from modules import tune_report

            recs = await asyncio.to_thread(tune_report.build, self._solana)
            for rec in recs:
                if getattr(rec, "applyable", False) and rec.key:
                    proposals.append((
                        int(getattr(rec, "rank", 50)), rec.key, str(rec.value),
                        None, rec.title,
                        "restart" in (getattr(rec, "caveat", "") or "").lower(),
                    ))
        except Exception as exc:  # noqa: BLE001
            logger.debug("[Heal] tune unavailable: %s", exc)

        # 2 — blockers that are settings rather than facts about the market.
        # Deliberately NOT auto-proposing DRY_RUN or SOLANA_EXECUTION_ENABLED:
        # those two spend real money the moment they flip, and a "heal"
        # button that offers to start live trading is a button someone taps
        # by accident once.
        try:
            blockers = self._solana._execution_blockers() if self._solana else []
            for b in blockers:
                if "DRY_RUN" in b or "EXECUTION_ENABLED" in b:
                    proposals.append((
                        99, "", "", None,
                        "execution is OFF — this one is yours to turn on, "
                        "deliberately, not from a heal button:\n"
                        "`./scripts/setenv.sh DRY_RUN=false "
                        "SOLANA_EXECUTION_ENABLED=true`",
                        False,
                    ))
                    break
        except Exception as exc:  # noqa: BLE001
            logger.debug("[Heal] blockers unreadable: %s", exc)

        # 3 — v1.89 — THE BLOCKER /heal COULD NOT SEE.
        #
        # The operator, after a week of this: "the system still not healing
        # him self good need more better".
        #
        # They are right, and this is the gap. /heal only ever looked at
        # tune_report's recommendations — which are all `KEY=value` settings
        # — plus DRY_RUN. Custody was never in it. So for a week the single
        # thing blocking every trade, printed at the top of /doctor, named
        # correctly by /verify and holding $268.74 of near-misses in /gates,
        # was absent from the one screen whose entire job is "here is what
        # to fix".
        #
        # It cannot be fixed with a setenv, and that was the unexamined
        # assumption: /heal was built for things it could propose as a y/n
        # card, so a blocker with no env var to change simply fell out of
        # the model. A repair screen that can only see repairs of one SHAPE
        # is not a repair screen, it is a settings editor.
        #
        # It still does not apply anything — this is an IAM grant against
        # AWS and the box correctly cannot make it. But it now leads with
        # the command, because the operator runs this from a phone and the
        # difference between "you have a permissions problem" and a
        # pasteable line is the whole distance to it being fixed.
        try:
            from modules import key_custody

            st = key_custody.status()
            if not st.get("loaded") and st.get("error"):
                err = str(st["error"])
                # NAMES THE INFRASTRUCTURE, so it goes only to the real
                # owner — not merely to "whoever passed the command gate".
                #
                # TELEGRAM_OWNER_OPEN=true opens every non-public command to
                # every chat, and this message carries the secret id, the
                # role name and the policy name: a map of what to attack and
                # what it is called. The rest of /heal proposes settings; it
                # is the only entry that hands out infrastructure, so it is
                # the only one that needs a tighter gate than the command it
                # arrives in. Found in review.
                #
                # _OWNER_CHAT_IDS is checked DIRECTLY, deliberately ignoring
                # _OWNER_GATE_OPEN. A flag that widens access to reports
                # must not also widen access to reconnaissance.
                owner_only = chat_id in _OWNER_CHAT_IDS
                if ("AccessDenied" in err or "IAM" in err) and owner_only:
                    secret = os.getenv("SOLANA_KEY_SECRET_ID", "").strip() \
                        or "garden-angel/solana-signer"
                    proposals.append((
                        100, "", "", None,
                        "🔑 *CANNOT SIGN — this is the top blocker and it is "
                        "not a setting.*\n\n"
                        "The instance role has no "
                        "`secretsmanager:GetSecretValue` on "
                        f"`{secret}`. I cannot grant it: this box correctly "
                        "cannot edit its own IAM, which is why "
                        "`fix_iam.sh --apply` is denied.\n\n"
                        "*Fastest path from a phone — no terminal:*\n"
                        "AWS Console → IAM → Roles → `SSMInstanceRole` → "
                        "Add permissions → Create inline policy → JSON tab → "
                        "paste, name it `garden-angel-read-signer-secret`, "
                        "Create.\n\n"
                        "`bash scripts/fix_iam.sh` prints the exact JSON and "
                        "the ARN with your account id already filled in.\n\n"
                        "_No restart needed after — custody is read per "
                        "attempt, so the next scan cycle picks it up._",
                        False,
                    ))
        except Exception as exc:  # noqa: BLE001
            logger.debug("[Heal] custody unreadable: %s", exc)

        actionable = [p for p in proposals if p[1]]
        advisory = [p for p in proposals if not p[1]]

        if not actionable and not advisory:
            await self._safe_send(
                "🩹 *nothing to heal*\n\nNo setting is currently misconfigured "
                "in a way I can fix with one key. If the bot still is not "
                "trading, that is the market or a blocker you control — "
                "/next ranks what is left.", chat_id)
            return

        # Highest priority first. Custody (100) leads, because a tuning
        # suggestion applied while the bot cannot sign is a change nobody
        # can measure — and measuring one change at a time is the rule this
        # screen closes with.
        advisory.sort(key=lambda p: -p[0])
        for note in advisory:
            await self._safe_send(f"⚠️ {note[4]}", chat_id)

        if not actionable:
            return

        actionable.sort(key=lambda p: -p[0])
        await self._safe_send(
            f"🩹 found *{len(actionable)}* fixable setting(s). Each needs its "
            f"own yes — nothing changes until you tap.", chat_id)
        for _prio, key, value, _unused, title, needs_restart in actionable[:6]:
            await self._propose_change(
                chat_id, key, value, title, "heal",
                needs_restart=needs_restart,
            )

    async def _cmd_verify(self, chat_id: int) -> None:
        """Can this bot actually trade, right now? One card, custody-aware.

        v1.71 — the operator has been reading
        scripts/push_key_to_secrets.sh's output as a verification and being
        told `FAILED — SOLANA_PRIVATE_KEY is not set in .env` on a box whose
        key correctly lives in AWS. That script is a one-way PUSH tool; it
        cannot verify anything, because it only knows whether a plaintext
        copy exists.

        This asks the questions that actually decide whether a signal can
        become a trade, through the same modules the send path uses.
        """
        from modules import key_custody

        lines = ["🧿 *verify*", ""]

        # 1 — signing.
        try:
            st = await asyncio.to_thread(key_custody.status)
            source = st.get("source") or "env"
            if st.get("loaded") and st.get("pubkey"):
                lines.append(f"✅ signs as `{st['pubkey']}`")
                lines.append(f"    custody: `{source}`")
            else:
                # Not yet resolved is not the same as broken — force one real
                # load so this answers the question it was asked.
                try:
                    await asyncio.to_thread(key_custody.load_private_key)
                    st = await asyncio.to_thread(key_custody.status)
                    lines.append(f"✅ signs as `{st.get('pubkey')}`")
                    lines.append(f"    custody: `{source}`")
                except Exception as exc:  # noqa: BLE001
                    lines.append(f"❌ CANNOT SIGN — {str(exc)[:180]}")
                    lines.append(f"    custody: `{source}`")
            if st.get("risk"):
                lines.append(f"⚠️ {st['risk']}")
            if st.get("mismatch"):
                lines.append(f"⚠️ {st['mismatch']}")
        except Exception as exc:  # noqa: BLE001
            lines.append(f"❌ custody unreadable: {str(exc)[:160]}")

        # 2 — is execution even switched on?
        lines.append("")
        try:
            blockers = self._solana._execution_blockers() if self._solana else []
            if blockers:
                lines.append("⛔ execution is HELD:")
                for b in blockers[:4]:
                    lines.append(f"    • {b}")
            else:
                lines.append("✅ execution is enabled")
        except Exception as exc:  # noqa: BLE001
            lines.append(f"· execution state unreadable: {str(exc)[:120]}")

        # 3 — the push cache, new in v1.71 and worth seeing.
        try:
            from modules.ws_state import get_ws_state

            ws = get_ws_state().status()
            if not ws["enabled"]:
                lines.append("· push cache off (SOLANA_WS_STATE=false)")
            elif ws.get("live"):
                lines.append(
                    f"✅ push cache live — {ws['slots_seen']:,} slots, "
                    f"{ws['blockhash_pushes']:,} blockhash pushes, "
                    f"{ws['alt_subscriptions']} ALT(s) watched"
                    + (f", last slot {ws['secs_since_slot']:.1f}s ago"
                       if ws.get("secs_since_slot") is not None else ""))
            elif ws["connected"]:
                # v1.86 — CONNECTED IS NOT LIVE.
                #
                # This branch used to be folded into the one above, reading
                # `connected` — True from the moment the TLS handshake
                # completes. A provider that accepts the socket and delivers
                # nothing (the exact shape of a missing or wrong auth token)
                # printed a green tick with a frozen slot count beside it,
                # while every trade paid the cold fetch this cache exists to
                # remove.
                lines.append(
                    f"🔴 push cache CONNECTED BUT NOT DELIVERING — "
                    f"{ws.get('stale_reason', 'no slots arriving')}. "
                    f"{ws['slots_seen']:,} slots seen in total. Every trade "
                    f"is paying the cold blockhash fetch right now.")
            else:
                lines.append(
                    f"⚠️ push cache not connected"
                    + (f" — {ws['last_error'][:100]}" if ws["last_error"] else "")
                    + " (harmless: the cache falls back to fetching)")
        except Exception:  # noqa: BLE001
            pass

        # 4 — gRPC readiness, only when it is being pursued. Silent otherwise:
        # a box happily running the WebSocket push cache should not be shown a
        # row of red crosses for a feature it has not asked for.
        try:
            from modules import grpc_preflight

            if grpc_preflight.status()["configured"]:
                lines.append("")
                lines.append(grpc_preflight.render())
        except Exception:  # noqa: BLE001
            pass

        # 5 — the gate that has been declining everything.
        try:
            from modules.solana_executor import _MIN_REAL_NET_MARGIN
            from modules.tip_memory import remembered_floor

            floor_lam, floor_age = await asyncio.to_thread(remembered_floor)
            sol_usd = (self._solana._last_sol_usd_price if self._solana else 0) or 0
            if floor_lam and sol_usd > 0:
                tip_usd = floor_lam / 1e9 * sol_usd
                lines.append("")
                lines.append(
                    f"💰 an edge must clear ~${tip_usd:.4f} tip + "
                    f"${_MIN_REAL_NET_MARGIN:.2f} margin = "
                    f"*${tip_usd + _MIN_REAL_NET_MARGIN:.4f}* to be worth sending")
                lines.append(
                    f"    _tip floor remembered {floor_age / 60:.0f}m ago_")
        except Exception:  # noqa: BLE001
            pass

        await self._safe_send("\n".join(lines), chat_id)

    async def _cmd_quote(self, chat_id: int, args: list) -> None:
        """A live round trip for any pair and size, from the phone.

        v1.71 — /quote existed and was removed. It is back because the
        question it answers ("is there anything there right now?") otherwise
        requires SSH, and the operator has been running scan scripts by hand
        on a box they reach from a phone.

        Read-only. Quotes only. No key is loaded, nothing is signed.
        """
        base = (args[0].upper() if len(args) > 0 else "USDC")
        mid = (args[1].upper() if len(args) > 1 else "SOL")
        try:
            size = float(args[2]) if len(args) > 2 else 1000.0
        except (TypeError, ValueError):
            await self._safe_send(
                "size must be a number — `/quote USDC SOL 1000`", chat_id)
            return

        from constants import SOLANA_TOKENS

        if base not in SOLANA_TOKENS or mid not in SOLANA_TOKENS:
            known = ", ".join(sorted(SOLANA_TOKENS)[:14])
            await self._safe_send(
                f"unknown token. Try: {known}\n\nUsage: `/quote USDC SOL 1000`",
                chat_id)
            return

        await self._safe_send(
            f"quoting {base}->{mid}->{base} at {size:,.0f} {base}…", chat_id)
        try:
            text = await self._round_trip_quote(base, mid, size)
        except Exception as exc:  # noqa: BLE001
            text = f"quote failed: {str(exc)[:220]}"
        await self._safe_send(text, chat_id)

    async def _round_trip_quote(self, base: str, mid: str, size: float) -> str:
        import httpx

        from constants import (JUPITER_HEADERS, JUPITER_QUOTE_API,
                               SOLANA_TOKENS)

        b, m = SOLANA_TOKENS[base], SOLANA_TOKENS[mid]
        amount_in = int(size * (10 ** b["decimals"]))

        async def _q(in_mint: str, out_mint: str, amount: int):
            r = await client.get(
                f"{JUPITER_QUOTE_API}/quote", headers=JUPITER_HEADERS, timeout=15.0,
                params={"inputMint": in_mint, "outputMint": out_mint,
                        "amount": str(amount), "slippageBps": "50",
                        "swapMode": "ExactIn"})
            r.raise_for_status()
            return r.json()

        async with httpx.AsyncClient() as client:
            q1 = await _q(b["mint"], m["mint"], amount_in)
            q2 = await _q(m["mint"], b["mint"], int(q1["outAmount"]))

        out = int(q2["outAmount"])
        gross_bps = (out - amount_in) / amount_in * 10_000.0
        gross_usd = (out - amount_in) / (10 ** b["decimals"])

        from modules.env_file import effective_haircut_bps

        haircut, _src = effective_haircut_bps()
        fee_bps = haircut + float(os.getenv("SOLANA_FLASH_FEE_BPS", "0.5") or 0.5)
        net_bps = gross_bps - fee_bps

        verdict = "✅ positive" if net_bps > 0 else "❌ negative"
        return "\n".join([
            f"📉 *{base}->{mid}->{base}* at {size:,.0f} {base}",
            "",
            f"round trip: `{gross_bps:+.2f}` bps  (${gross_usd:+.4f})",
            f"fees:       `{fee_bps:.2f}` bps",
            f"net:        `{net_bps:+.2f}` bps   {verdict}",
            "",
            f"_both legs quoted live at this size, both impacts paid._",
        ])

    async def _cmd_next(self, chat_id: int) -> None:
        """What should I do next? One ranked answer.

        v1.71 — operator: "I want the system guide us to what we go next."

        Everything needed for this already existed and was scattered across
        /why, /doctor, /gates, /tune, /pipeline and /tips — six screens, each
        correct, none of which says which one to read first. This ranks.
        """
        items: list[tuple[int, str, str]] = []   # (priority, headline, action)

        # 1 — can it sign and is execution on? Nothing else matters if not.
        try:
            from modules import key_custody

            st = await asyncio.to_thread(key_custody.status)
            if st.get("error"):
                items.append((100, "the bot cannot load its signing key",
                              "/verify shows the custody error"))
            if st.get("plaintext_env_key_present") and st.get("source") != "env":
                items.append((90, "a plaintext key is still in .env beside the vault",
                              "remove SOLANA_PRIVATE_KEY from .env"))
        except Exception:  # noqa: BLE001
            pass

        try:
            blockers = self._solana._execution_blockers() if self._solana else []
            if blockers:
                items.append((95, "execution is switched off, so no signal can trade",
                              f"/why — {blockers[0][:90]}"))
        except Exception:  # noqa: BLE001
            pass

        # 2 — is the edge simply too small to be worth landing? This is the
        # finding the whole deployment has been circling for weeks.
        try:
            from modules.solana_executor import _MIN_REAL_NET_MARGIN
            from modules.tip_memory import remembered_floor

            floor_lam, _age = await asyncio.to_thread(remembered_floor)
            sol_usd = (self._solana._last_sol_usd_price if self._solana else 0) or 0
            if floor_lam and sol_usd > 0:
                need = floor_lam / 1e9 * sol_usd + _MIN_REAL_NET_MARGIN
                items.append((
                    60,
                    f"an edge must be worth ${need:.2f}+ to be worth sending",
                    "if signals are smaller than this, the answer is a bigger "
                    "edge, not a bigger tip — /observe, /hunt, discover.py",
                ))
        except Exception:  # noqa: BLE001
            pass

        # 3 — is the pipeline fast enough to keep what it finds?
        try:
            from modules.pipeline_trace import get_log

            summary = await asyncio.to_thread(get_log().summary)
            p90 = (summary or {}).get("total_ms_p90")
            if p90 and p90 > 400:
                items.append((
                    70,
                    f"the pipeline p90 is {p90:.0f}ms — over one 400ms slot",
                    "/pipeline names the slowest stage; a quote older than a "
                    "slot is now refused before it is built",
                ))
        except Exception:  # noqa: BLE001
            pass

        # 4 — anything /tune wants, as a proposal rather than an instruction.
        try:
            from modules import tune_report

            recs = await asyncio.to_thread(tune_report.build, self._solana)
            top = next((r for r in recs if r.applyable), None)
            if top is not None:
                items.append((50, f"/tune suggests: {top.title}",
                              "run /tune and approve it if you agree"))
        except Exception:  # noqa: BLE001
            pass

        if not items:
            await self._safe_send(
                "🧠 *next*\n\nNothing is blocking and nothing is asking for a "
                "decision. The bot is scanning; if it is not trading, that is "
                "the market, not the machine. /observe for the honest view.",
                chat_id)
            return

        items.sort(key=lambda t: -t[0])
        lines = ["🧠 *next* — most important first", ""]
        for n, (_prio, headline, action) in enumerate(items[:5], 1):
            lines.append(f"*{n}. {headline}*")
            lines.append(f"    {action}")
            lines.append("")
        await self._safe_send("\n".join(lines), chat_id)

    async def _cmd_pipeline(self, chat_id: int) -> None:
        """Where the time between signal and wire actually goes."""
        from modules.pipeline_report import render_pipeline

        text = await asyncio.to_thread(render_pipeline)
        await self._safe_send(text, chat_id)

    async def _cmd_lighthouse(self, chat_id: int) -> None:
        """One verdict from all eleven screens (v17.49).

        Operator: "we are at darkness we need go to light."

        Every screen this bot has is correct and together they are a
        reading assignment. Each answers a different question and none
        answers THE question — is there money here, and if not, what is the
        one next thing to do. This ranks them and names one action.
        """
        from modules.lighthouse_report import render_lighthouse

        text = await asyncio.to_thread(render_lighthouse, self._solana)
        await self._safe_send(text, chat_id)

    async def _probe_jupiter_venues(self) -> list[tuple[str, bool, str]]:
        """Price 1 SOL → USDC through each venue via Jupiter's `dexes=`.

        This is the venue access the bot actually has. The direct adapters
        below it are a fallback for a Jupiter outage, and two of the three
        no longer have an endpoint to fall back to — so probing only those
        reported the bot as blind to Whirlpool and Meteora while it was
        quoting both every cycle.
        """
        import httpx

        from constants import JUPITER_HEADERS, JUPITER_QUOTE_API, SOLANA_TOKENS

        sol, usdc = SOLANA_TOKENS["SOL"], SOLANA_TOKENS["USDC"]
        venues = ["Whirlpool", "Orca V2", "Meteora DLMM", "Raydium",
                  "Raydium CLMM", "Phoenix"]
        # v1.77 — Lifinity V2 removed from the probe list.
        #
        # It answers HTTP 400 to a plain 1-SOL quote and has done at every
        # probe. Lifinity runs an oracle-based market maker, so it declines
        # to quote pairs/sizes outside its configured bands rather than
        # returning a worse price — a 400 there is the venue working as
        # designed, not a fault.
        #
        # Printing it in red on every /venues taught the operator to scroll
        # past red, which is the one thing a status screen must never do.
        # Same reasoning as retiring the orca/meteora adapters: an alarm
        # that fires forever for something that cannot be fixed is not
        # information.
        #
        # Jupiter still ROUTES through Lifinity whenever it is the best
        # path — this list is only which venues get an isolated price probe.
        out: list[tuple[str, bool, str]] = []

        async with httpx.AsyncClient(timeout=15.0) as client:
            for venue in venues:
                try:
                    r = await client.get(
                        f"{JUPITER_QUOTE_API}/quote",
                        params={"inputMint": sol["mint"], "outputMint": usdc["mint"],
                                "amount": str(10 ** sol["decimals"]),
                                "slippageBps": "50", "dexes": venue,
                                "onlyDirectRoutes": "true"},
                        headers=JUPITER_HEADERS,
                    )
                    if r.status_code != 200:
                        out.append((venue, False, f"HTTP {r.status_code}"))
                        continue
                    raw = r.json().get("outAmount")
                    if not raw:
                        # A venue with no direct pool for this pair is a
                        # real answer, not an error — it is why the
                        # cross-venue scan discards some legs as illiquid.
                        out.append((venue, False, "no direct route for this pair"))
                        continue
                    price = int(raw) / (10 ** usdc["decimals"])
                    out.append((venue, True, f"1 SOL → {price:,.2f} USDC"))
                except Exception as exc:  # noqa: BLE001
                    out.append((venue, False, str(exc)[:60]))
                await asyncio.sleep(0.3)
        return out

    async def _cmd_sendpath(self, chat_id: int) -> None:
        """The 400ms test, on demand, without needing a signal (v17.56).

        Operator: "i need this test alone in telegram."

        /pipeline can only refresh when a real signal reaches the executor.
        /bench times stages in isolation on a fresh connection, which
        measures the network rather than the caches. This calls the exact
        functions a live send calls, through the same hot_state cache, cold
        then warm — so the warm column is what a trade actually pays.
        """
        import asyncio as _a

        from constants import SOLANA_TOKENS
        from modules.solana_executor import (
            _cached_alt_accounts, _cached_mint_safety, _cached_priority_fee,
            _cached_reserve, _fresh_blockhash, get_shared_client,
        )

        rpc = (os.getenv("SOLANA_RPC_PRIMARY")
               or os.getenv("SOLANA_RPC_URL") or "").strip()
        if not rpc:
            await self._safe_send("⏱ *sendpath* — no RPC configured.", chat_id)
            return

        client = get_shared_client()
        sol, usdc = SOLANA_TOKENS["SOL"], SOLANA_TOKENS["USDC"]
        stages = [
            ("freeze SOL", lambda: _cached_mint_safety(client, rpc, "SOL", sol["mint"])),
            ("freeze USDC", lambda: _cached_mint_safety(client, rpc, "USDC", usdc["mint"])),
            ("priority fee", lambda: _cached_priority_fee(client, rpc)),
            ("reserve", lambda: _cached_reserve(client, "USDC", usdc["mint"], rpc)),
            ("alt tables", lambda: _cached_alt_accounts(client, rpc, [])),
            ("blockhash", lambda: _fresh_blockhash(client, rpc)),
        ]

        async def _t(fn):
            t0 = time.monotonic()
            try:
                r = fn()
                if _a.iscoroutine(r):
                    await r
            except Exception as exc:  # noqa: BLE001
                return None, str(exc)[:44]
            return (time.monotonic() - t0) * 1000.0, ""

        sign = os.getenv("GA_SIGN", "✝️🪬🧿 🧠🦅👀")
        out = [f"{sign} ⏱ *sendpath* — the 400ms test", "",
               "_the real cached calls a send makes. nothing is signed._", "",
               "`stage            COLD    WARM`"]
        warm_total = 0.0
        broken = []
        for name, fn in stages:
            cold, err = await _t(fn)
            if cold is None:
                broken.append(f"{name}: {err}")
                out.append(f"`{name:<14} FAILED`")
                continue
            warms = []
            for _ in range(3):
                w, _e = await _t(fn)
                if w is not None:
                    warms.append(w)
            if not warms:
                broken.append(f"{name}: warm call failed")
                continue
            warm = sorted(warms)[len(warms) // 2]
            warm_total += warm
            out.append(f"`{name:<14}{cold:>6.0f}ms{warm:>6.0f}ms`")

        if broken:
            out += ["", "⚠️ *some stages failed — this is not a pass:*"]
            out += [f"  · {b}" for b in broken]
            await self._safe_send("\n".join(out), chat_id)
            return

        quotes = 120.0
        total = warm_total + quotes
        out += ["",
                f"cached state, warm  `{warm_total:>6.0f} ms`",
                f"+ quotes & swap-ix  `{quotes:>6.0f} ms`  _(the market, uncacheable)_",
                f"*= signal to signed* `{total:>6.0f} ms`",
                f"a Solana slot is    `{400:>6.0f} ms`", ""]
        out.append(
            f"✅ *{total:.0f} ms — inside one slot.* The warm cache is working; "
            f"latency is not what stops a trade. /observe for whether the round "
            f"trip is positive at all."
            if total <= 400 else
            f"🔴 *{total:.0f} ms — longer than a slot.* A warm column over 100ms "
            f"means that cache is missing — check it restarted and that "
            f"warm_mint_safety ran."
        )
        await self._safe_send("\n".join(out), chat_id)

    async def _cmd_tips(self, chat_id: int) -> None:
        """What the tip auction actually cost, remembered (v17.53).

        Operator: "save as memory to the bot". /jito reports the engine's
        in-RAM counters, which reset on every restart — several times an
        hour on this box — so its "0/0 landed" was a claim about uptime,
        not about the auction. This reads modules/tip_memory.py, which
        persists every bid, the floor it faced, and its fate.
        """
        from modules.tip_memory import render_tips

        text = await asyncio.to_thread(render_tips)
        await self._safe_send(text, chat_id)

    async def _cmd_observe(self, chat_id: int) -> None:
        """Was this trade EVER viable, over recorded history (v17.50).

        Every other screen answers a question about right now. This one
        reads the observatory's stored series and answers the question that
        decides whether to keep going at all — see modules/observatory.py
        for why latency is deliberately absent from the arithmetic.
        """
        from modules.observatory import render_observe

        text = await asyncio.to_thread(render_observe)
        await self._safe_send(text, chat_id)

    async def _cmd_bench(self, chat_id: int) -> None:
        """Per-stage micro-benchmark of the SEND path (v17.48).

        /speedtest answers "are the endpoints up" and has been answering
        ✅ PASS for weeks while nothing traded, because endpoint health and
        send-path speed are different questions. This times each stage a
        send actually uses and composes them into the one number that
        decides a race: quote → signed, against Solana's 400ms slot.
        """
        from modules.bench_report import run_bench

        await self._safe_send("⏱ benchmarking the send path… ~15s", chat_id)
        text = await run_bench(self._solana)
        await self._safe_send(text, chat_id)

    async def _cmd_decisions(self, chat_id: int) -> None:
        """Why each EVALUATION did not become a trade — the 99.9% every
        other screen structurally cannot see (v17.47).

        Operator, showing their Rust bot's equivalent: a quarter of a
        million rows reading `skip:negative_gross`. That table is not
        noise, it is the finding — it says the constraint sits upstream of
        every gate this bot argues about. /capture, /nearmiss and /gates
        all start from a signal, so none of them can say anything about
        the cycles that never produced one.
        """
        from modules.pipeline_report import render_decisions

        text = await asyncio.to_thread(render_decisions)
        await self._safe_send(text, chat_id)

    async def _cmd_gates(self, chat_id: int) -> None:
        """Which of our OWN rules stopped a trade, and what they held."""
        from modules.pipeline_report import render_gates

        text = await asyncio.to_thread(render_gates)
        await self._safe_send(text, chat_id)

    async def _cmd_drawdown(self, chat_id: int) -> None:
        """Drawdown, loss streak, 24h net, and the alert latches (v17.39)."""
        from modules.drawdown_alert import get_watcher

        watcher = get_watcher()
        st = watcher.status()
        cur = st["current"] or {}
        if not cur:
            # No poll has run yet — do one now rather than reporting
            # "unknown" for a figure that is one file read away.
            try:
                await asyncio.to_thread(watcher.check)
                st = watcher.status()
                cur = st["current"] or {}
            except Exception as exc:  # noqa: BLE001
                logger.debug("[/drawdown] on-demand check failed: %s", exc)

        th = st["thresholds"]
        lines = ["📉 *drawdown watch*\n"]
        if not st["enabled"]:
            lines.append("🔴 disabled (DRAWDOWN_ALERT_ENABLED=false)")
            await self._safe_send("\n".join(lines), chat_id)
            return

        if not cur or not cur.get("executed"):
            lines.append(
                "No executed trades yet, so every figure here is trivially zero. "
                "Alerts stay silent until there is a realised curve to draw down "
                "from — a $0.00 drawdown on 0 trades is not a clean record."
            )
        else:
            lines.append(
                f"drawdown: *-${cur['drawdown_usd']:,.2f}* "
                f"(alerts at ${th['drawdown_usd']:,.2f})"
            )
            lines.append(
                f"net realised: ${cur['net_profit_usd']:,.2f} over "
                f"{cur['executed']:,} trade(s)"
                + (f", win rate {cur['win_rate']:.0%}" if cur.get("win_rate") is not None else "")
            )
            lines.append(
                f"losing streak: *{cur['loss_streak']}* "
                f"(alerts at {th['loss_streak']})"
            )
            lines.append(
                f"last 24h: ${cur['day_net_usd']:,.2f} over "
                f"{cur['day_trades']:,} trade(s) "
                f"(alerts at -${th['day_loss_usd']:,.2f})"
            )

        armed = [n for n, l in st["latches"].items() if l["armed"]]
        fired = [n for n, l in st["latches"].items() if not l["armed"]]
        lines.append(
            f"\nlatches: {', '.join(armed) or 'none'} armed"
            + (f" · {', '.join(fired)} already fired" if fired else "")
        )
        lines.append(
            f"halt on breach: "
            f"{'ON — execution pauses too' if st['halt_on_breach'] else 'off (alert only)'}"
        )
        lines.append(
            "\n_Each alert fires once on crossing, again only if it gets "
            f"${th['step_usd']:,.2f} worse, and re-arms when the metric halves. "
            "An alert that repeats every poll is one you mute._"
        )
        await self._safe_send("\n".join(lines), chat_id)

    # ── /doctor — plain-language "can it trade, and if not, why" (v17.28) ─────
    #
    # Modeled on the operator's separate Rust bot's /doctor (see module
    # changelog). Read-only: every check reads state the engine already
    # keeps, plus at most one best-effort getBalance RPC — never a Jupiter
    # call, so this can be run as often as the operator likes without
    # spending rate-limit budget or perturbing an in-flight scan cycle.

    async def _cmd_doctor(self, chat_id: int) -> None:
        problems: list[str] = []   # blocks trading outright, or breaks it
        warnings: list[str] = []   # temporary / non-fatal
        info: list[str] = []       # settings + state worth seeing at a glance

        sol = self._solana
        if sol is None:
            problems.append(
                "Solana engine is not attached — the scan loop never started "
                "(check the server startup log)."
            )
        else:
            try:
                problems.extend(sol._execution_blockers())
            except Exception as exc:
                warnings.append(f"could not read execution blockers: {str(exc)[:100]}")

            if sol.scan_count == 0:
                warnings.append(
                    "no scan cycle has completed yet — startup warm-up is ~45s, "
                    "give it a minute"
                )
            else:
                age = time.time() - sol.last_scan_ts if sol.last_scan_ts else None
                if age is not None and age > 180:
                    problems.append(
                        f"scan loop looks STALE — last completed cycle was {age:.0f}s "
                        "ago (expected every ~20-40s); a restart may be needed"
                    )
                else:
                    info.append(
                        f"scan loop alive — last cycle {age:.0f}s ago, "
                        f"{sol.scan_count} cycles, {sol.signal_count} signal(s) this run"
                    )

            try:
                if sol._jupiter_is_cooling():
                    remaining = max(0.0, sol._jupiter_backoff_until - time.monotonic())
                    warnings.append(
                        f"Jupiter is rate-limited (cooling {remaining:.0f}s more) — "
                        "scanning and execution resume by themselves after"
                    )
                else:
                    info.append("Jupiter: not rate-limited right now")
            except Exception:
                pass

            # v17.36 — the trading circuit breaker is the single most likely
            # reason a correctly-configured bot is not executing, so it goes
            # near the top of the problem list rather than buried in info.
            try:
                from modules.trading_guard import get_guard

                guard = get_guard().status()
                if guard["state"] == "open":
                    problems.append(
                        f"trading is AUTO-PAUSED by the circuit breaker — "
                        f"{guard['trip_reason'][:220]}. Retries by itself in "
                        f"{guard['cooldown_remaining_secs'] / 60:.1f} min, or "
                        f"/resume to clear now. (Scanning is unaffected.)"
                    )
                elif guard["state"] == "half_open":
                    warnings.append(
                        "trading circuit breaker is HALF-OPEN — the next execution "
                        "attempt is a trial run; if it succeeds, normal trading resumes"
                    )
                elif guard["consecutive"]:
                    # v17.41 — only warn about streaks that can actually
                    # trip something. A threshold of 0 means "never trips"
                    # (edge_decayed is deliberately set that way: declining
                    # an unprofitable trade is the bot working, not a
                    # fault), so "streak building: edge_decayed×11" reads
                    # as an alarm about correct behaviour and trains the
                    # operator to ignore this whole section. The count is
                    # still worth having, just not as a warning — it moves
                    # down to state, without the raw threshold dict that
                    # was being printed at a phone screen verbatim.
                    thresholds = guard["thresholds"] or {}
                    live = {
                        kind: n for kind, n in guard["consecutive"].items()
                        if thresholds.get(kind, 0) > 0
                    }
                    inert = {
                        kind: n for kind, n in guard["consecutive"].items()
                        if thresholds.get(kind, 0) <= 0
                    }
                    if live:
                        warnings.append(
                            "consecutive failure streak building: "
                            + ", ".join(
                                f"{k}×{v} (trips at {thresholds[k]})"
                                for k, v in live.items()
                            )
                        )
                    if inert:
                        info.append(
                            "counted, cannot pause trading: "
                            + ", ".join(f"{k}×{v}" for k, v in inert.items())
                            + " — a correct refusal is not a fault"
                        )
            except Exception as exc:  # noqa: BLE001
                warnings.append(f"could not read the trading guard: {str(exc)[:100]}")

            # v17.36 — key custody. A plaintext key is not a trade blocker,
            # so it is a warning, not a problem — but it is the one warning
            # here that can cost the whole wallet, so it always shows.
            try:
                from modules import key_custody

                custody = key_custody.status()
                if custody.get("mismatch"):
                    problems.append(custody["mismatch"])
                if custody.get("risk"):
                    # v17.45 — the operator has read this warning many times
                    # and asked not to see it again. Two honest ways to grant
                    # that: migrate the key (the risk goes away and so does
                    # the line), or acknowledge it (the risk stays, the
                    # nagging stops).
                    #
                    # A warning that repeats unchanged on every /doctor stops
                    # being read — including by the person it is for — and
                    # then it is worse than useless, because it trains the
                    # eye to skip the whole ⚠️ block where a NEW warning will
                    # one day appear. So: acknowledged demotes it to a single
                    # state line. It is never removed, because the wallet is
                    # still bearer-authority with no revocation, and quietly
                    # deleting that fact is not something a bot that signs
                    # transactions should do.
                    if _env_bool("GA_ACK_PLAINTEXT_KEY", False):
                        info.append(
                            "key custody: PLAINTEXT (acknowledged) — still "
                            "bearer authority, no revocation. `bash "
                            "scripts/push_key_to_secrets.sh` when you want it "
                            "gone for real."
                        )
                    else:
                        warnings.append(
                            f"key custody — {custody['risk']} "
                            f"Fix: `bash scripts/push_key_to_secrets.sh`. "
                            f"Or silence this line with "
                            f"`./scripts/setenv.sh GA_ACK_PLAINTEXT_KEY=true` "
                            f"— the risk stays, the reminder stops."
                        )
                else:
                    info.append(
                        f"key custody: {custody['source']}"
                        + (f", signs for {custody['pubkey']}" if custody.get("pubkey") else "")
                    )
            except Exception as exc:  # noqa: BLE001
                warnings.append(f"could not read key custody: {str(exc)[:100]}")

            # v1.83 — A FAILOVER THAT CANNOT FAIL OVER.
            #
            # This screen counted a configured SOLANA_RPC_SECONDARY as
            # redundancy without ever comparing it to the primary. Two URLs
            # on one provider read as two endpoints everywhere and behave as
            # one: the 401 that killed the primary kills the secondary in
            # the same instant, because it is the same account.
            #
            # Env only, no network — /doctor's promise is that it costs
            # nothing to run, and this keeps it.
            try:
                from modules.rpc_failover import failover_health

                fh = failover_health()
                problems.extend(fh["problems"])
                warnings.extend(fh["warnings"])
                if fh["has_secondary"] and not fh["same_host"] and not fh["same_domain"]:
                    info.append(
                        f"RPC failover: {fh['primary_host']} → "
                        f"{fh['secondary_host']} — genuinely separate "
                        f"providers, so one outage does not take both"
                    )
            except Exception as exc:  # noqa: BLE001
                warnings.append(f"could not check RPC failover: {str(exc)[:100]}")

            # v1.83 — A FEED THAT IS ON AND BUYING NOTHING.
            #
            # feed_client falls back to Jupiter on every uncertainty, by
            # design — a wrong "skip" costs a trade. The cost of that
            # correctness is that a feed reading the wrong key looks
            # EXACTLY like a quiet market: no errors, no warnings, a healthy
            # Redis, and a Jupiter budget that never drops. The whole point
            # of the feed is the budget dropping.
            #
            # Only runs when FEED_ENABLED=true, so it costs nothing today.
            try:
                from modules.feed_client import ENABLED as _feed_on, get_feed

                if _feed_on:
                    fd = get_feed().describe()
                    # Only a KEY MISMATCH is a blocker — see describe()'s own
                    # note. An unreachable Redis or an absent writer is the
                    # documented state until garden-feed exists, and filing
                    # it under ⛔ put "trading is unaffected" directly above
                    # the one entry that genuinely blocks trading.
                    for p in fd.get("problems", []):
                        problems.append(f"pool feed: {p}")
                    if fd.get("hint"):
                        problems.append(f"    fix: `{fd['hint']}`")
                    for w in fd.get("warnings", []):
                        warnings.append(w)
                    if fd.get("info"):
                        info.append(fd["info"])
            except Exception as exc:  # noqa: BLE001
                warnings.append(f"could not check the pool feed: {str(exc)[:100]}")

            # v1.87 — the shared Jupiter bucket. Same silent-failure shape
            # as the feed: it falls open on every error by design, so a
            # broken limiter and a working one produce identical trading
            # behaviour and identical silence — the difference only ever
            # shows up as a 429 nobody can explain.
            try:
                from modules.jupiter_limiter import get_limiter

                jl = get_limiter().describe()
                problems.extend(jl["problems"])
                if jl["info"]:
                    info.append(jl["info"])
            except Exception as exc:  # noqa: BLE001
                warnings.append(f"could not check the Jupiter limiter: {str(exc)[:100]}")

            # v17.36 — route pruning and the dynamic floor both change what
            # counts as a signal, so "why did it stop reporting X" needs
            # them visible here.
            try:
                from modules.route_scorer import get_scorer

                scoring = get_scorer().status()
                # v17.43 — THE BLACKOUT CHECK.
                #
                # On 2026-07-31 this section printed "route pruning: 9 of 13
                # routes paused" as ordinary info, under the heading "✅ no
                # problems found — configuration can trade", while /hunt
                # reported 13 of 13 not quoted and 0 calls. The scanner had
                # been off the air for 355 cycles and every screen in the
                # bot described that as tidy optimisation.
                #
                # A cycle that issues no quotes cannot find a signal. There
                # is no configuration in which that is acceptable, so it is
                # a PROBLEM, stated first, in those words.
                if scoring.get("blackout"):
                    problems.append(
                        f"SCANNER IS DARK — {scoring.get('blackout_reason', 'no routes scanned')}. "
                        f"No quotes means no signals, however healthy everything else "
                        f"looks. Fix order: (1) `./scripts/setenv.sh ROUTE_FOCUS_TOP_N=0` "
                        f"to stop concentrating, (2) `./scripts/setenv.sh "
                        f"ROUTE_SCORE_PRUNING_ENABLED=false` if it persists, "
                        f"(3) /routes to see which rule excluded what."
                    )
                planned, total = scoring.get("planned_scan_count"), scoring.get("planned_total")
                if planned is not None and total:
                    frac = scoring.get("budget_fraction")
                    budget = (
                        f"{scoring.get('budget_used', 0):,}/{scoring.get('budget_total', 0):,} "
                        f"Jupiter calls used this hour"
                        + (f" ({frac * 100:.0f}%)" if frac is not None else "")
                    )
                    if not scoring.get("rationing"):
                        info.append(
                            f"scan set: ALL {planned} of {total} route(s) — {budget}, "
                            f"below the {scoring.get('budget_pressure_on', 0.6) * 100:.0f}% "
                            f"line where concentrating starts to pay for itself. "
                            f"Nothing is being skipped to save a budget that is idle."
                        )
                    else:
                        info.append(
                            f"scan set: {planned} of {total} route(s) — {budget}, "
                            f"so the budget is being concentrated on the producers "
                            f"(/routes for who and why)"
                        )
                # v17.43 — a route excluded for lacking evidence it could not
                # have produced is a different thing from a dead one, and
                # conflating them is what pruned the best route in the book.
                if scoring.get("routes_undersampled"):
                    info.append(
                        f"{scoring['routes_undersampled']} route(s) UNDERSAMPLED — not "
                        f"enough evaluations yet for '0 signals' to mean anything at a "
                        f"{scoring.get('min_detectable_rate', 0.002) * 100:.2f}% signal "
                        f"rate. They keep scanning; /routes shows how many more each needs."
                    )
                if scoring.get("routes_proven"):
                    info.append(
                        f"{scoring['routes_proven']} route(s) PROVEN — they have cleared "
                        f"the floor at least once, so nothing may prune them however "
                        f"quiet the last few hundred evals were"
                    )
                if scoring.get("rescued"):
                    # v17.44 — THIS ASKED A QUESTION THE BOT HAD ALREADY
                    # ANSWERED, ONCE PER CYCLE, FOREVER.
                    #
                    # The text offered two possibilities — "a market fact"
                    # or "the thresholds are too tight" — and left the
                    # operator to pick. But /observe has been pricing real
                    # cross-venue moments against the real tip floor for
                    # days, and its verdict is not ambiguous: over 1,600
                    # measurements, never once viable. That settles it. The
                    # floor is binding because the market is not paying.
                    #
                    # The survival floor doing its job is not a fault
                    # condition. It exists precisely so a flat market cannot
                    # prune the scanner down to nothing, and warning every
                    # time the safeguard works is warning about the design.
                    # Same class as /tune advising a value already in force:
                    # a line that costs attention and returns nothing, in a
                    # list that is only useful if everything in it is worth
                    # reading.
                    #
                    # So: consult the evidence. Settled by measurement ->
                    # say what was measured, in info. No evidence, or
                    # evidence that says otherwise -> the question is still
                    # open and the warning still earns its place.
                    rescued_txt = ", ".join(scoring["rescued"])
                    try:
                        from modules.observatory import settled_unprofitable

                        verdict = settled_unprofitable()
                    except Exception:  # noqa: BLE001
                        verdict = None
                    if verdict is not None:
                        info.append(
                            f"survival floor holding {rescued_txt} open — "
                            f"working as designed. /observe priced "
                            f"{verdict['priced']:,} real moments over "
                            f"{verdict.get('hours', 0):.0f}h and none cleared "
                            f"full cost, so this is the market, not the "
                            f"thresholds. Nothing to change here."
                        )
                    else:
                        warnings.append(
                            "the survival floor is binding — pruning and focus "
                            "together would have left too few routes scanning, "
                            f"so {rescued_txt} were kept regardless. Either "
                            "everything is genuinely unprofitable right now (a "
                            "market fact) or the thresholds are too tight — "
                            "/observe has not priced enough moments to say "
                            "which."
                        )
                if scoring["routes_pruned"]:
                    info.append(
                        f"route pruning: {scoring['routes_pruned']} of "
                        f"{scoring['routes_scored']} routes paused for never coming "
                        f"near the floor (/routes for the list) — their Jupiter "
                        f"budget goes to the routes that do produce"
                    )
            except Exception:  # noqa: BLE001 — diagnostics, never a blocker
                pass

            # The config knobs that have actually decided "trade vs no trade"
            # in this deployment's history, with the one known-fatal
            # misconfiguration checked explicitly.
            try:
                effective = sol.effective_min_profit()
                if effective > sol.min_profit * 1.05:
                    # v1.83 — NAME THE GATE THAT ACTUALLY RAISED IT.
                    #
                    # This line blamed "priority-fee volatility" for every
                    # raise, unconditionally, because effective_min_profit()
                    # returns one float and there was nothing to attribute
                    # it with. On this deployment it reported a $1.50 floor
                    # over a $0.05 config and blamed fees — while the fee
                    # gate caps at static × MAX_MULT = $0.30 and literally
                    # cannot produce $1.50, and /decay on the next screen
                    # was naming the real $1.454 out loud. /why then told
                    # the operator to set MIN_PROFIT_FLOOR_USD to the value
                    # it already had. See min_profit_breakdown()'s docstring.
                    try:
                        bd = sol.min_profit_breakdown()
                        parts = []
                        if bd["dynamic_adder"] > 0.005:
                            parts.append(
                                f"+${bd['dynamic_adder']:.2f} priority-fee "
                                f"volatility (DYNAMIC_FLOOR_ENABLED)"
                            )
                        if bd["decay_adder"] > 0.005:
                            parts.append(
                                f"+${bd['decay_adder']:.2f} edge-decay gate "
                                f"(DECAY_GATE_ENABLED) — /decay"
                            )
                        cause = " · ".join(parts) or "composed gates"
                        knob = {
                            "decay": "DECAY_GATE_ENABLED=false removes it "
                                     "(it is priced from 20 attempts that all failed)",
                            "dynamic": "it drops back by itself in a quiet moment",
                            "bps": "MIN_PROFIT_FLOOR_BPS is what binds, not the dollar floor",
                        }.get(bd["dominant"], "")
                        info.append(
                            f"signal floor: net ≥ ${effective:.2f} right now = "
                            f"${sol.min_profit:.2f} configured {cause}. "
                            f"Raising MIN_PROFIT_FLOOR_USD is NOT what set this"
                            + (f" — {knob}" if knob else "")
                        )
                    except Exception:  # noqa: BLE001
                        info.append(
                            f"signal floor: net ≥ ${effective:.2f} right now "
                            f"(configured ${sol.min_profit:.2f}, raised by a "
                            f"composed gate — /decay and /tune for which)"
                        )
                else:
                    info.append(f"signal floor: net ≥ ${effective:.2f} after fees")
                # v17.46 — the PERCENTAGE floor, and what it works out to.
                #
                # effective_min_profit() is called here with no loan size, so
                # the bps floor cannot apply and this line reported $0.03
                # while the real bar at a 2,500 probe was $0.50 — sixteen
                # times higher. A floor line that names one of two floors is
                # the same failure as "learned flash-fee floor: 3.9 bps" over
                # a file that could not be written: technically true, and it
                # sends the operator to tune the wrong number.
                try:
                    from modules.solana_arb import _MIN_PROFIT_FLOOR_BPS

                    if _MIN_PROFIT_FLOOR_BPS > 0:
                        probe = float(os.getenv("SOLANA_PROBE_BASE_UNITS", "") or 10_000)
                        usd = probe * sol._usd_per_base_unit("USDC")
                        as_dollars = usd * _MIN_PROFIT_FLOOR_BPS / 10_000.0
                        binding = "THIS is the binding floor" if as_dollars > effective \
                            else "the dollar floor above is higher, so it binds"
                        info.append(
                            f"percentage floor: {_MIN_PROFIT_FLOOR_BPS:.2f} bps of the "
                            f"loan = ${as_dollars:.2f} at the current {probe:,.0f} probe "
                            f"— {binding}. Unlike a dollar floor this does not change "
                            f"meaning when the probe size does."
                        )
                except Exception:  # noqa: BLE001
                    pass
            except Exception:  # noqa: BLE001
                info.append(f"signal floor: net ≥ ${sol.min_profit:.2f} after fees")
            # v17.55 — ASK FOR THE VALUE THAT IS ACTUALLY APPLIED.
            #
            # This read SOLANA_LEG2_HAIRCUT_BPS directly, which is the one
            # thing modules/env_file.py's effective_haircut_bps() exists to
            # prevent — its docstring says so in as many words: "Scripts
            # must not read SOLANA_LEG2_HAIRCUT_BPS directly any more. With
            # the adaptive governor on, that variable is only a fallback,
            # and quoting it at the operator invites tuning a number that
            # no longer does anything."
            #
            # /doctor was doing exactly that, and it cost the operator real
            # time. Their learned haircut is 0.30 bps; this read the dead
            # 2.0 fallback, and the test below is `haircut > |best| + 1.0`.
            # With best = -0.9 that is 2.0 > 1.9 — a PROBLEM raised on every
            # /doctor, telling them to "Set SOLANA_LEG2_HAIRCUT_BPS=2"
            # which is the value it had just misread. With the real 0.30 it
            # never fires.
            #
            # The source is now shown alongside the number, so a fallback
            # can never again be mistaken for the enforced value.
            try:
                from modules.env_file import effective_haircut_bps

                haircut, haircut_src = effective_haircut_bps()
            except Exception:  # noqa: BLE001
                haircut, haircut_src = float(
                    os.getenv("SOLANA_LEG2_HAIRCUT_BPS")
                    or os.getenv("SOLANA_SOL_WRAP_HAIRCUT_BPS")
                    or 2.0
                ), "configured"
            # v17.38 (2026-07-29 — operator's /doctor reported "leg-2 drift
            # buffer: 5.0 bps" as ordinary state while the bot sat at 0
            # signals across 1,093 cycles) — the old test was `> 5.0`, so
            # exactly 5.0 read as fine. It is not fine, and a fixed
            # threshold was the wrong instrument anyway: what makes a
            # haircut fatal is not its absolute size, it is its size
            # RELATIVE TO THE EDGES THIS DEPLOYMENT ACTUALLY SEES. The
            # journal knows those. Compare against them instead of against
            # a number someone picked in 2026-07.
            best_bps = None
            try:
                from modules.route_scorer import get_scorer

                spreads = [
                    r["mean_spread_bps"] for r in get_scorer().table() if r["rows"] > 0
                ]
                if spreads:
                    best_bps = max(spreads)
            except Exception:  # noqa: BLE001 — fall back to the fixed test
                best_bps = None

            if best_bps is not None and haircut > abs(best_bps) + 1.0:
                # The remedy names the RIGHT lever. When the applied value
                # is learned, setting the env var changes nothing — it is
                # only the fallback — so telling an operator to set it is
                # sending them to do nothing and then wonder why /doctor
                # still complains.
                fix = (
                    "The learned governor is what applies it — "
                    "`./scripts/setenv.sh HAIRCUT_GOVERNOR_ENABLED=false "
                    f"SOLANA_LEG2_HAIRCUT_BPS={min(2.0, max(0.1, abs(best_bps) / 2)):.1f}` "
                    "pins it instead. /decay shows what the governor is buying."
                    if haircut_src == "learned" else
                    f"Set SOLANA_LEG2_HAIRCUT_BPS="
                    f"{min(2.0, max(0.1, abs(best_bps) / 2)):.1f} and restart."
                )
                problems.append(
                    f"leg-2 haircut is {haircut:.2f} bps ({haircut_src}), but the best "
                    f"route in your own trade journal averages {best_bps:+.1f} bps. The "
                    f"haircut is bigger than the entire edge — every re-quote is "
                    f"arithmetically certain to come back short, which is exactly why "
                    f"signals fire and nothing ever sends. {fix}"
                )
            elif haircut >= 5.0:
                problems.append(
                    f"leg-2 haircut is {haircut:.1f} bps — larger than almost any real "
                    "edge these routes produce (0.1-3 bps), so every re-quote fails the "
                    "margin check. This exact misconfiguration froze all trading before "
                    "2026-07-27; the fixed default is 2.0. Set "
                    "SOLANA_LEG2_HAIRCUT_BPS=2 in .env and restart."
                )
            else:
                # v17.42 — say whether this number is still in force. With
                # the governor on, /doctor listed "leg-2 drift buffer: 2.0
                # bps" and "learned haircut: 0.30 bps" as two peers, and
                # nothing said which one the executor obeys. Given this
                # deployment's history — twice frozen by exactly this
                # constant — an ambiguous haircut reading is the single
                # worst line to leave on the page.
                adaptive_on = False
                try:
                    from modules.haircut_governor import ENABLED as _HG_ON

                    adaptive_on = bool(_HG_ON)
                except Exception:  # noqa: BLE001
                    pass
                if adaptive_on:
                    info.append(
                        f"leg-2 drift buffer: {haircut:.1f} bps configured — "
                        "NOT in use, the learned value below is what BOTH the "
                        "scanner and the executor apply (fallback only, if the "
                        "governor is disabled or unreadable). Before v17.43 the "
                        "scanner used this static number while the executor used "
                        "the learned one, so the scan was pricing a worse trade "
                        "than the one it was screening for — on 0.1-3 bps routes, "
                        "most of the signal."
                    )
                else:
                    info.append(
                        f"leg-2 drift buffer: {haircut:.1f} bps"
                        + (f" (best observed edge {best_bps:+.1f} bps)" if best_bps else "")
                    )
            # The learned haircut, which is what the executor actually uses
            # when adaptive mode is on. Reported separately from the static
            # setting above so the two can never be confused for each other.
            try:
                from modules.haircut_governor import get_governor

                hg = get_governor().status()
                if hg["enabled"]:
                    quiet = hg["hours_since_shortfall"]
                    info.append(
                        f"learned haircut: {hg['bps']:.2f} bps "
                        f"({hg['raises']} raise(s), {hg['decays']} decay(s)"
                        + (f", quiet {quiet:.1f}h" if quiet is not None else "")
                        + ") — rises on a real 6024, decays back down when "
                        "none appear. /decay for what it is buying."
                    )
                    if hg["at_ceiling"]:
                        problems.append(
                            f"learned haircut is pinned at its {hg['ceiling_bps']:.1f} "
                            "bps ceiling — leg 1 keeps delivering far less than it "
                            "quotes. That is not ordinary drift: suspect the route, "
                            "the slippage setting, or a stale reserve. Raising the "
                            "ceiling would just recreate the paralysis."
                        )
            except Exception:  # noqa: BLE001 — diagnostics only
                pass

            # v17.38 — the decay measurement, once there is one. This is the
            # direct answer to "signals fire but nothing sends".
            try:
                from modules.decay_tracker import get_tracker

                decay = get_tracker().status()
                if decay["samples"] >= decay["min_samples"] and decay["decay_usd_per_sec"]:
                    info.append(
                        f"edge decay: ${decay['decay_usd_per_sec']:.4f}/s over a "
                        f"{decay['pipeline_p90_secs']:.2f}s pipeline (p90) — a signal "
                        f"needs ${decay['implied_adder_usd']:.3f} of headroom above the "
                        f"floor just to survive to send. /decay for the breakdown"
                    )
                elif decay["recording"]:
                    info.append(
                        f"edge decay: {decay['samples']}/{decay['min_samples']} attempts "
                        f"recorded — /decay once there are enough to measure"
                    )
            except Exception:  # noqa: BLE001 — diagnostics only
                pass
            # v17.44 — the flash fee, and whether the number above is
            # actually REMEMBERED.
            #
            # This line read "learned flash-fee floor: 3.9 bps" for days
            # while the file behind it could not be written (it pointed at
            # /data/, which does not exist on this box) and the failure was
            # logged at DEBUG. So every restart forgot, assumed a flash loan
            # was free, announced a +0.5 bps signal against a real 2.94 bps
            # fee, and burned it in simulation — over and over. A status
            # line that cannot tell "proven and durable" from "in memory
            # until the next restart" is a status line that lies by
            # omission, so it now checks the write path instead of assuming
            # it. See solana_arb.learned_flash_fee_status().
            try:
                from modules.solana_arb import learned_flash_fee_status

                fee_state = learned_flash_fee_status()
                if not fee_state["durable"]:
                    problems.append(
                        f"the learned flash-fee floor CANNOT BE SAVED "
                        f"({fee_state['detail']}). Every restart forgets what the "
                        f"last one proved on-chain, so the same unprofitable signal "
                        f"gets found, announced and burned again — forever. Fix: "
                        f"`./scripts/setenv.sh SOLANA_FLASH_FEE_STATE_FILE=data/"
                        f"flash_fee_learned.json`"
                    )
                if sol.flash_fee_bps <= 0:
                    problems.append(
                        "the flash-loan fee is assumed to be ZERO. It is not — a "
                        "flash loan is a service. Assuming it is free is what turns "
                        "a +0.5 bps quote into an announced signal that simulation "
                        "then rejects. Set SOLANA_FLASH_FEE_BPS=3 or higher."
                    )
                else:
                    per = fee_state.get("by_reserve") or {}
                    proven = (
                        " · proven per reserve: "
                        + ", ".join(f"{k[:8]}…={v:.2f}" for k, v in sorted(per.items()))
                        if per else ""
                    )
                    info.append(
                        f"flash-fee floor: {sol.flash_fee_bps:.2f} bps "
                        f"(hard floor {fee_state['floor_bps']:.2f}, learned "
                        f"{fee_state['learned_bps']:.2f}){proven} — "
                        f"{'remembered across restarts' if fee_state['durable'] else 'NOT DURABLE'}"
                    )
            except Exception:  # noqa: BLE001 — diagnostics never block
                if sol.flash_fee_bps > 0:
                    info.append(
                        f"learned flash-fee floor: {sol.flash_fee_bps:.2f} bps "
                        "(proven on-chain; raises the bar on every route)"
                    )

            # v17.44 — the noise gate's own tally. A high reject rate is not
            # a fault: it is the count of 🦅-then-❌ pairs that no longer
            # happen. Worth seeing so the operator knows it is working
            # rather than wondering where the signals went.
            try:
                from modules.noise_gate import status as noise_status

                ng = noise_status()
                if ng["checked"]:
                    info.append(
                        f"noise gate: {ng['rejected']} of {ng['checked']} candidate "
                        f"signal(s) held back for being inside their own scatter. "
                        f"Each one would have announced and then failed its "
                        f"pre-send re-quote."
                    )
            except Exception:  # noqa: BLE001
                pass
            loan_override = os.getenv("SOLANA_REAL_LOAN_UNITS", "").strip()
            if loan_override:
                info.append(
                    f"real loan override: {loan_override} units (still capped to "
                    "each signal's own proven size)"
                )

            # ── v17.43 — grade the OUTCOME, not only the configuration ────
            #
            # On 2026-07-31 every check above passed and this screen said
            # "✅ no problems found — configuration can trade" while the
            # scanner was dark and 17 of 17 attempts had landed nothing.
            # Both statements were true. The configuration WAS valid. A
            # configuration check structurally cannot notice that nothing
            # is happening, and presenting it as the summary invited
            # exactly the wrong conclusion.
            #
            # These four look at results. /why ranks them properly by where
            # in the funnel they bite; this is the short version, so the two
            # screens can never disagree about whether there is a problem.
            try:
                from modules.capture_report import capture

                cap = await asyncio.to_thread(capture)
                attempted = cap.get("attempted") or 0
                landed = cap.get("landed") or 0
                signalled = cap.get("signalled") or 0
                if attempted >= 10 and landed == 0:
                    problems.append(
                        f"{attempted} execution attempts, {landed} landed. Every "
                        f"one died between the signal and a confirmed "
                        f"transaction. /why ranks the cause; /pipeline says "
                        f"where the time goes; /gates says how many were "
                        f"stopped by our own rules rather than by the market."
                    )
                elif signalled >= 10 and attempted == 0:
                    problems.append(
                        f"{signalled} signals found and not one was ever "
                        f"attempted — something is refusing before the "
                        f"executor is reached. /gates names it."
                    )
            except Exception:  # noqa: BLE001 — diagnostics never block
                pass

            try:
                from modules.pipeline_trace import get_log as _pipe_log

                pipe = _pipe_log().status()
                p50 = pipe.get("total_p50_ms")
                # v17.54 — do not raise a PROBLEM from a stale window.
                #
                # A pipeline trace is only written when a signal reaches the
                # executor. When nothing sends, the last window sits frozen
                # — the operator's has read 1.19s over 3 attempts for four
                # days, measured before the freeze warm-cache fix, on code
                # that no longer exists.
                #
                # /pipeline and /lighthouse were taught this in #156;
                # /doctor is the fifth screen to read the same number and
                # the last one still treating it as current. An operator
                # asked for "the doctor clean" is owed a screen that does
                # not send them after a stage that was fixed days ago and
                # cannot report otherwise until something sends.
                if p50 and pipe.get("stale"):
                    age_d = (pipe.get("newest_age_secs") or 0) / 86400.0
                    info.append(
                        f"pipeline: last traced {age_d:.1f}d ago at "
                        f"{p50 / 1000.0:.2f}s — too old to judge, and it "
                        f"cannot refresh until a signal reaches the executor. "
                        f"Not counted as a problem. `scripts/test_sendpath.py` "
                        f"measures the send path without needing a signal."
                    )
                elif p50 and p50 > 1500:
                    slowest = pipe.get("slowest") or "?"
                    problems.append(
                        f"the pipeline takes {p50 / 1000.0:.2f}s from signal to "
                        f"wire, and the slowest stage is `{slowest}`. On edges "
                        f"that live for seconds that is most of the reason a "
                        f"re-quote comes back short. /pipeline for the split."
                    )
                elif p50:
                    info.append(
                        f"pipeline: {p50 / 1000.0:.2f}s p50 signal → wire "
                        f"(slowest stage `{pipe.get('slowest')}`) — /pipeline"
                    )
            except Exception:  # noqa: BLE001
                pass

            # The control arm's verdict, when it has one. This is the single
            # line that decides whether latency work is worth paying for,
            # and it must never be buried — /nearmiss has been asserting
            # "that is a race you are losing" without the evidence to say so.
            try:
                from modules.decay_control import get_control

                control = get_control().status()
                if control.get("verdict") not in (None, "off", "not yet"):
                    info.append(
                        f"decay control: {control['verdict']} — "
                        f"{control.get('explanation', '')[:300]}"
                    )
            except Exception:  # noqa: BLE001
                pass

            # The single most-asked question — what happened to the LAST
            # signal — answered from the same exec_detail the dashboard shows.
            for opp in reversed(sol.last_opportunities or []):
                detail = getattr(opp, "exec_detail", "") or ""
                if getattr(opp, "is_signal", False) and detail:
                    info.append(f"last signal outcome: {detail[:350]}")
                    break

            # ── Fee-balance check (v17.37 — REWRITTEN) ────────────────────
            #
            # Operator report, 2026-07-29: Phantom showed 0.07375 SOL
            # ($5.45) in the wallet while /doctor reported "signing wallet
            # holds 0 lamports (0.0000 SOL)". Both were telling the truth
            # about DIFFERENT ADDRESSES.
            #
            # The bug: this check read SOLANA_WALLET. That variable is
            # documented in solana_arb.resolve_wallet_address() as a
            # "reference address, no signing key derived" — a fallback used
            # only when no private key is available. The address that
            # actually pays transaction fees is the one derived from
            # SOLANA_PRIVATE_KEY, and resolve_wallet_address() has
            # preferred it for exactly that reason since v1.2. /doctor was
            # the one place that ignored it and checked the reference
            # address instead. Fund one Phantom account, point
            # SOLANA_WALLET at another (or leave an old value in place) and
            # /doctor confidently reports an empty wallet forever.
            #
            # Made worse by the message never printing the address it
            # checked, so there was nothing in the output to compare
            # against the wallet app. That is fixed too — every balance
            # line now names the address, because "0 SOL" is only
            # actionable once you can see WHICH account it means.
            #
            # Now: check the SIGNER first (the account that pays), check
            # SOLANA_WALLET as well when it differs, and report the
            # divergence explicitly rather than silently preferring either.
            signer, signer_source = sol.resolve_wallet_address()
            wallet_env = os.getenv("SOLANA_WALLET", "").strip()

            async def _lamports(address: str) -> tuple[Optional[int], str]:
                """Balance via primary RPC, falling back to the secondary.

                A single-endpoint read was its own quiet failure mode here:
                a throttled or lagging RPC answers 0 for an account it
                hasn't indexed, which is indistinguishable from an empty
                wallet in the old message.
                """
                import httpx

                urls = [u for u in (sol.rpc_url, getattr(sol, "rpc_url_secondary", "")) if u]
                last = ""
                async with httpx.AsyncClient() as client:
                    for url in urls:
                        try:
                            r = await client.post(
                                url,
                                json={"jsonrpc": "2.0", "id": 1,
                                      "method": "getBalance", "params": [address]},
                                timeout=6.0,
                            )
                            body = r.json()
                            if isinstance(body, dict) and body.get("error"):
                                last = str(body["error"])[:100]
                                continue
                            return int(body["result"]["value"]), ""
                        except Exception as exc:  # noqa: BLE001 — try the next endpoint
                            last = f"{type(exc).__name__}: {str(exc)[:70]}"
                return None, last or "no RPC endpoint configured"

            checked: dict[str, Optional[int]] = {}
            if signer and sol.rpc_url:
                lamports, err = await _lamports(signer)
                checked[signer] = lamports
                if lamports is None:
                    warnings.append(
                        f"could not read the signing wallet's balance ({err}) — "
                        f"address `{signer}`. Not a trade blocker by itself, but "
                        f"nothing below could be verified either."
                    )
                elif lamports < 3_000_000:
                    problems.append(
                        f"the signing wallet holds {lamports:,} lamports "
                        f"({lamports / 1e9:.4f} SOL) — not enough to pay "
                        f"transaction fees, so every send fails at simulation. "
                        f"Top up ~0.01 SOL.\n"
                        f"  address: `{signer}`\n"
                        f"  ({signer_source})\n"
                        f"  ⚠️ check that address against your wallet app before "
                        f"sending anything — if your funded account shows a "
                        f"DIFFERENT address, the bot is signing with the wrong key "
                        f"and topping this one up will not help."
                    )
                else:
                    info.append(
                        f"fee wallet: {lamports / 1e9:.4f} SOL — enough for fees "
                        f"(`{signer[:8]}…{signer[-6:]}`, {signer_source})"
                    )
            elif not signer:
                # v17.44 — name the custody source ACTUALLY in use.
                #
                # This text sent the operator to SOLANA_PRIVATE_KEY, an
                # environment variable they deliberately removed when the
                # key moved to AWS Secrets Manager ("i dont want the bot
                # read the key from env its not secure"). On a vault-backed
                # box the advice was not merely useless, it pointed at the
                # thing that is supposed to be missing — and the real fault
                # would be in the vault, unmentioned.
                try:
                    from modules import key_custody as _kc

                    src = _kc.active_source() or "env"
                except Exception:  # noqa: BLE001
                    src = "env"
                where = ("SOLANA_PRIVATE_KEY" if src == "env"
                         else f"the {src} signing key")
                warnings.append(
                    f"no signing address could be resolved — {where} could not "
                    f"be read and SOLANA_WALLET is not usable either, so the "
                    f"fee-balance check was skipped. `/verify` says why."
                )

            # The divergence itself. This is the actual root cause of the
            # 2026-07-29 report, so it is stated in full rather than left
            # for the operator to infer from two addresses in two places.
            if wallet_env and signer and wallet_env != signer:
                env_lamports, _ = await _lamports(wallet_env)
                checked[wallet_env] = env_lamports
                env_desc = (
                    f"{env_lamports / 1e9:.4f} SOL" if env_lamports is not None else "unreadable"
                )
                signer_lamports = checked.get(signer)
                signer_desc = (
                    f"{signer_lamports / 1e9:.4f} SOL"
                    if signer_lamports is not None else "unreadable"
                )
                message = (
                    f"SOLANA_WALLET and the actual signing key are DIFFERENT "
                    f"addresses:\n"
                    f"  signs with:    `{signer}` — {signer_desc}\n"
                    f"  SOLANA_WALLET: `{wallet_env}` — {env_desc}\n"
                    f"Fees are paid by the SIGNER. Whichever of these your wallet "
                    f"app shows as funded is the one that matters"
                )
                if signer_lamports is not None and signer_lamports < 3_000_000 \
                        and env_lamports is not None and env_lamports >= 3_000_000:
                    # The exact reported case: money in one account, signing
                    # from another. Not a warning — trading cannot work.
                    problems.append(
                        message + " — and right now the funded one is NOT the "
                        "signer, which is why nothing can trade. Either set "
                        "SOLANA_PRIVATE_KEY to the key for the funded account, or "
                        "move ~0.01 SOL to the signer's address above."
                    )
                else:
                    warnings.append(
                        message + ". Set SOLANA_WALLET to the signer's address to "
                        "stop the dashboard and this check disagreeing."
                    )
            elif wallet_env and not signer:
                info.append(f"SOLANA_WALLET: `{wallet_env}` (no signing key derived)")

        # v17.48 — /doctor MUST NOT say "no problems" while /why is showing
        # a red. Operator: "dont let the doctor hide the warn i see in
        # other".
        #
        # They caught a real inconsistency. This screen checks CONFIGURATION
        # and /why checks OUTCOME, so for weeks the two disagreed in the
        # worst possible direction: ✅ here, 🔴 there, and the ✅ is the one
        # that reads like a summary. Two screens that contradict each other
        # do not give the operator a choice of opinions — they teach them
        # that neither can be trusted.
        #
        # So /why's blocking findings are pulled in directly. Not re-derived
        # (that would be a third opinion to keep in sync), just carried
        # across, attributed, and pointed back at the screen that explains
        # them properly.
        try:
            from modules.why_report import build as _why_build

            # v17.57 — DEDUPE ON MEANING, NOT ON A PREFIX SUBSTRING.
            #
            # The old test was `_f.headline[:40] in existing`, and the
            # operator's /doctor duly printed the same fact twice:
            #
            #   • 20 execution attempts, 0 landed. Every one died between…
            #   • 20 attempts, 0 landed — every one died between…
            #
            # "20 attempts, 0 landed" is not a SUBSTRING of "20 execution
            # attempts, 0 landed" — the word `execution` sits in the middle
            # — so the guard passed and /why's copy was appended anyway. A
            # screen that reports two problems when it has found one
            # inflates its own count and makes the list look worse than the
            # box is.
            #
            # Comparing the distinctive words instead is robust to that kind
            # of rewording. Numbers are kept because "20 attempts" and "3
            # attempts" are different findings; short filler words are
            # dropped because they carry no identity.
            _STOP = {"the", "a", "an", "and", "of", "in", "to", "is", "was",
                     "it", "on", "at", "by", "for", "with", "that", "every",
                     "one", "not", "no", "its", "this", "are", "from"}

            def _sig(text: str) -> set:
                import re as _re

                return {w for w in _re.findall(r"[a-z0-9.$]+", text.lower())
                        if w not in _STOP and len(w) > 1}

            _seen = [_sig(p) for p in problems]
            for _f in _why_build(sol).blocking():
                _line = f"{_f.headline} — {_f.evidence[:160]}" if _f.evidence else _f.headline
                _s = _sig(_f.headline)
                # Two thirds of the distinctive words already present means
                # this is the same finding in different words.
                _dupe = any(
                    _s and len(_s & prior) / len(_s) >= 0.66 for prior in _seen
                )
                if not _dupe:
                    problems.append(f"{_line} (/why for the full picture)")
                    _seen.append(_s)
        except Exception as exc:  # noqa: BLE001 — never break /doctor
            logger.debug("[CommandHandlers] /why cross-check failed: %s", exc)

        lines = ["🩺 doctor — Solana leg\n"]
        if problems:
            lines.append(f"⛔ {len(problems)} problem(s):")
            lines.extend(f"• {p}" for p in problems)
        else:
            lines.append("✅ no problems found — configuration can trade.")
            lines.append(
                "(a signal still only sends when its fresh re-quote clears the "
                "margin at send time — a quiet stretch means thin edges, not a fault)"
            )
        if warnings:
            lines.append("")
            lines.append("⚠️ worth knowing:")
            lines.extend(f"• {w}" for w in warnings)
        if info:
            lines.append("")
            lines.append("ℹ️ state:")
            lines.extend(f"• {i}" for i in info)
        await self._tg.send_message("\n".join(lines), chat_id=chat_id, parse_mode=None)

    # ── /netdiag — on-demand IPv4/IPv6 connectivity probe (v17.15) ────────────
    #
    # Runs in the background like /hunt: diagnose_network() does up to
    # 5 hosts × 2 address families × a 5s connect timeout each, so a
    # fully-broken network could take ~50s worst case. Acking immediately
    # and spawning the probe keeps the dispatcher responsive to other
    # commands (e.g. /status) while it runs, instead of blocking handle()
    # for the full duration.

    async def _cmd_netdiag(self, chat_id: int) -> None:
        if diagnose_network is None:
            await self._tg.send_message(
                "⚠️ /netdiag unavailable — modules/netdiag.py is not deployed "
                "in this environment.",
                chat_id=chat_id, parse_mode=None,
            )
            return

        await self._safe_send(
            "🔍 Running network diagnostic (IPv4 + IPv6, up to ~50s worst case)…",
            chat_id,
        )
        self._spawn(self._run_netdiag(chat_id))

    async def _run_netdiag(self, chat_id: int) -> None:
        try:
            results = await diagnose_network()
        except Exception as exc:
            logger.error("[CommandHandlers] /netdiag failed: %s", exc, exc_info=True)
            await self._safe_send(f"⚠️ Diagnostic failed: {str(exc)[:300]}", chat_id)
            return

        if not results:
            await self._safe_send("Diagnostic returned no results.", chat_id)
            return

        lines = ["🌐 *Network Diagnostic*\n"]
        for r in results:
            ipv4 = r.get("ipv4")
            ipv6 = r.get("ipv6")
            ipv4_ok = bool(ipv4 and ipv4[0])
            ipv6_ok = bool(ipv6 and ipv6[0])

            if ipv4_ok and ipv6_ok:
                verdict = "✅ both OK"
            elif ipv4_ok and not ipv6_ok and ipv6:
                verdict = "⚠️ IPv4 OK / IPv6 broken"
            elif ipv6_ok and not ipv4_ok and ipv4:
                verdict = "⚠️ IPv6 OK / IPv4 broken"
            elif ipv4 or ipv6:
                verdict = "❌ both failed"
            else:
                verdict = "— no records"

            lines.append(f"`{r.get('host', '?')}`\n  {verdict}")

        await self._tg.send_message("\n".join(lines), chat_id=chat_id, parse_mode="Markdown")

    # ── /speedtest — Solana RPC/WS + Jupiter latency diagnostic (v1.0) ────────
    #
    # Same "ack immediately, run in the background" shape as /netdiag above
    # — a probe round trip against 5 targets (RPC primary/secondary, WS
    # primary/secondary, Jupiter) shouldn't block the dispatcher. See
    # modules/speedtest.py's run_speedtest() for the shared probing logic
    # (also used by scripts/speedtest.py's CLI and the bot API's
    # "speedtest" action) and modules/dashboard_panels.py's
    # render_speedtest() for the shared markdown formatter both use.

    async def _cmd_journal(self, chat_id: int) -> None:
        """v1.35 — plain-language read of the trade journal. Reading and
        aggregating a CSV is blocking file I/O, so it runs in a thread
        rather than stalling the command loop once the file is large."""
        import asyncio
        from modules.journal_report import render_journal_report
        try:
            report = await asyncio.to_thread(render_journal_report)
        except Exception as exc:
            logger.error("[CommandHandlers] /journal failed: %s", exc, exc_info=True)
            await self._safe_send(f"⚠️ Journal read failed: {str(exc)[:300]}", chat_id)
            return
        await self._tg.send_message(report, chat_id=chat_id, parse_mode="Markdown")

    async def _cmd_speedtest(self, chat_id: int) -> None:
        await self._safe_send("⚡ Running speedtest (Solana RPC/WS + Jupiter)…", chat_id)
        self._spawn(self._run_speedtest(chat_id))

    async def _run_speedtest(self, chat_id: int) -> None:
        from modules import dashboard_panels as panels
        try:
            result = await panels.render_speedtest(self)
        except Exception as exc:
            logger.error("[CommandHandlers] /speedtest failed: %s", exc, exc_info=True)
            await self._safe_send(f"⚠️ Speedtest failed: {str(exc)[:300]}", chat_id)
            return
        await self._tg.send_message(result, chat_id=chat_id, parse_mode="Markdown")

    # ── v17.21 — AI copilot: /ai chat + /ai_idea → /approve flow ──────────────

    async def _cmd_ai(self, chat_id: int, args: list[str]) -> None:
        question = " ".join(args).strip()
        # v17.40 — BARE /ai IS A REPORT, NOT A USAGE MESSAGE.
        #
        # The operator asked twice for the Rust bot's `/ai`: hour-of-day
        # clearing rates and a landing model, both computed from that
        # bot's own history. Ours printed three example questions.
        #
        # This screen needs no API key, because nothing on it is
        # generated — see ai_model.py. That ordering matters: the learned
        # numbers must still be reachable on a box with no QWEN_API_KEY,
        # so the configuration check now guards only the chat path below.
        if not question:
            from modules.ai_model import render_model

            try:
                screen = render_model()
            except Exception as exc:  # noqa: BLE001
                logger.warning("[AI] learned screen failed: %s", str(exc)[:200])
                screen = ("🧠 could not build the learned screen "
                          f"({str(exc)[:120]}). Ask a question with "
                          "`/ai <question>`.")
            if self._qwen is None or not getattr(self._qwen, "enabled", False):
                screen += ("\n\n_Free-form questions need `QWEN_API_KEY`, "
                           "which is not set — the numbers above do not._")
            await self._safe_send(screen, chat_id)
            return
        if self._qwen is None or not getattr(self._qwen, "enabled", False):
            await self._safe_send(
                "🤖 AI is not configured — set QWEN_API_KEY in the Space "
                "secrets and restart.\n\nSend `/ai` with no question for the "
                "learned model, which needs no key.", chat_id,
            )
            return
        last = _recent_results[-1] if _recent_results else {}
        try:
            pair_count = len(getattr(self._scanner, "_pairs", []) or [])
        except Exception:
            pair_count = 0
        # v17.22 (2026-07-17, operator report — a real /ai answer suggested
        # lowering the profit floor to catch a "0.15-0.25% spread," but
        # that math only makes sense against a specific loan size, which
        # was never in the context the model saw — it had no way to check
        # its own claim against the real numbers. See qwen_client.py's own
        # matching v1.14 changelog entry for the prompt-side guardrail this
        # pairs with.
        #
        # v17.23 (2026-07-17, operator report — a real /ai answer cited
        # "flash-loan fees (typically 0.09% on Arbitrum, or $450)" — this
        # deployment's REAL configured rate, per the Scanner's own startup
        # log, is 0.05% ($250 on a $500k loan). loan_fee_pct was never in
        # this context either, same gap as loan_amount's own v17.22 fix —
        # the model had no real number to use and fell back to a generic
        # "typical Aave fee" figure that doesn't match this deployment.
        loan_amount = float(getattr(self._scanner, "loan_amount", 0) or 0)
        loan_fee_pct = float(getattr(self._scanner, "loan_fee_pct", 0) or 0)
        # v17.25 (2026-07-17, operator report — "🤖"-prefixed advice citing a
        # "$500k Solana size", a "$250 Solana fee", a "$1.50 floor cushion",
        # and Camelot fee-tier switches, none of which exist in this
        # deployment's config: the model had ZERO Solana-side numbers in
        # context, so everything Solana it said was invented, and the
        # loan_amount/fee here are ARB-only figures it happily reused for
        # Solana claims. Same gap-class as v17.22/v17.23, now closed for the
        # whole Solana leg: real floor, probe size, real-send fraction, fee
        # bps, and the latest scan's top route with its actual gap-to-floor
        # — the exact numbers its Solana suggestions need to be checkable
        # against. Best-effort: self._solana is None whenever the Solana leg
        # is off, and a summary failure must never break /ai itself.
        solana_ctx = ""
        if self._solana is not None:
            try:
                sol = self._solana
                sol_lines = [
                    f"Solana leg: floor ${float(sol.min_profit):.2f}/trade; "
                    f"scan probe size {float(sol.probe_base_units):,.0f} base units; "
                    f"REAL sends are capped to a small fraction of probe size "
                    f"(a signal at probe size can die at real size — "
                    f"'real size too thin'); "
                    f"flash fee {float(sol.flash_fee_bps):.1f} bps"
                ]
                opps = sol.last_opportunities or []
                if opps:
                    best = max(opps, key=lambda o: o.net_profit_usd)
                    sol_lines.append("; best route last scan: " + best.near_miss_line(float(sol.min_profit)))
                solana_ctx = "".join(sol_lines) + "; "
            except Exception:
                solana_ctx = ""
        context = (
            f"pairs watched: {pair_count}; ARB loan size per trade: ${loan_amount:,.2f}; "
            f"ARB flash-loan fee: {loan_fee_pct:.3f}% (this deployment's REAL "
            f"configured rate — do not substitute a generic/typical industry "
            f"figure; these two numbers are ARBITRUM-ONLY — never reuse them "
            f"for Solana claims); "
            + solana_ctx +
            f"last ARB scan: {last.get('signal', '?')} "
            f"{last.get('base', '?')} {last.get('buyOn', '?')}→{last.get('sellOn', '?')} "
            f"net ${float(last.get('netAfterFee', 0) or 0):.2f} "
            f"(floor ${float(last.get('minProfit', 0) or 0):.2f}); "
            f"ghost={self._ghost_mode} mint={self._mint_mode} dry_run={self._dry_run}"
        )
        answer = await self._qwen.ask(question, context=context)
        if answer:
            await self._safe_send("🤖 " + answer, chat_id)
            return
        reason = getattr(self._qwen, "last_error", None)
        await self._safe_send(
            "🤖 The AI didn't answer"
            + (f" ({reason})" if reason else "")
            + " — try again in a minute.",
            chat_id,
        )

    # v17.24 — /ai_idea used to accept the first candidate propose_pair()
    # returned as long as its address matched a 0x…40-hex REGEX, with no
    # check that the address was a real deployed contract. Retrying a few
    # times server-side (instead of making the operator keep resending
    # /ai_idea by hand) and grounding each candidate against the live chain
    # gives the AI a genuine shot at a usable suggestion per command.
    _AI_IDEA_MAX_ATTEMPTS = 3

    async def _cmd_ai_idea(self, chat_id: int) -> None:
        if self._qwen is None or not getattr(self._qwen, "enabled", False):
            await self._safe_send(
                "🤖 AI is not configured — set QWEN_API_KEY in the Space "
                "secrets and restart.", chat_id,
            )
            return
        await self._safe_send("🤖 Asking the AI for one concrete pair idea…", chat_id)
        try:
            watching = [
                f"{p['base']}/{p['stable']}"
                for p in (getattr(self._scanner, "_pairs", []) or [])
            ]
        except Exception:
            watching = []
        idea: dict | None = None
        verify: dict | None = None
        reject_reasons: list[str] = []
        for _attempt in range(self._AI_IDEA_MAX_ATTEMPTS):
            # v1.8 FIX (2026-07-15, "let the ai help the bot find the deal") —
            # this used to call propose_pair() with no chain, which internally
            # defaulted to BSC (PancakeSwap/Biswap/etc) — the wrong network
            # for this Arbitrum deployment (bot.py only wires w3_by_chain for
            # ARB), so an approved idea could never get a live on-chain
            # reserve read. Explicitly request the live chain now.
            candidate = await self._qwen.propose_pair(chain="ARB", already_watching=watching)
            if not candidate:
                reject_reasons.append(
                    getattr(self._qwen, "last_error", None)
                    or "model returned no usable suggestion"
                )
                continue
            # v17.24 (2026-07-17, operator report — "we need to make the ai
            # like agent suggestion and fix and help the bot it's still not
            # work" — two consecutive /ai_idea calls both failed with no
            # explanation) — propose_pair()'s address check was a REGEX
            # only; it passed a syntactically-valid but possibly nonexistent
            # or wrong address just as easily as a real one. Ground the
            # candidate against the live chain (scanner.py's new
            # verify_erc20_token()) before ever showing it to the operator.
            try:
                v = await self._scanner.verify_erc20_token(
                    str(candidate.get("chain", "ARB")), candidate["address"],
                )
            except Exception as exc:
                logger.warning("[CommandHandlers] verify_erc20_token failed: %s", exc)
                v = None
            if v is not None and not v.get("exists", False):
                reject_reasons.append(
                    f"{candidate.get('base', '?')} at "
                    f"{candidate['address'][:10]}… — {v.get('reason', 'not a real contract')}"
                )
                continue
            idea, verify = candidate, v
            break
        if not idea:
            detail = reject_reasons[-1] if reject_reasons else "no reason given"
            await self._safe_send(
                f"🤖 No usable suggestion after {self._AI_IDEA_MAX_ATTEMPTS} "
                f"tries ({detail}). Try /ai_idea again.",
                chat_id,
            )
            return
        reason = str(idea.pop("reason", "") or "")
        # Ground truth from the live chain wins over the AI's own guesses.
        verify_note = ""
        if verify and verify.get("exists"):
            if verify.get("decimals") is not None:
                idea["base_decimals"] = int(verify["decimals"])
            onchain_symbol = str(verify.get("symbol") or "").strip().upper()
            verify_note = "\n✅ Verified on-chain: contract exists"
            if onchain_symbol:
                verify_note += f", symbol={onchain_symbol}"
                if onchain_symbol != str(idea.get("base", "")).upper():
                    verify_note += (
                        f"\n⚠️ Mismatch: the AI called this '{idea.get('base')}' "
                        f"but the contract's own symbol() says '{onchain_symbol}' — "
                        "double-check this is really the token you think it is."
                    )
            verify_note += "."
        else:
            verify_note = (
                "\n⚠️ Could not verify on-chain right now (RPC unavailable) — "
                "check the explorer yourself before approving."
            )
        self._pending_ai_pair = idea
        explorer = {"ARB": "arbiscan.io", "BSC": "bscscan.com"}.get(
            str(idea.get("chain", "")).upper(), "the block explorer"
        )
        await self._safe_send(
            "🤖 AI pair idea — awaiting YOUR approval\n\n"
            f"Pair:  {idea['base']}/{idea['stable']} on {idea['chain']}\n"
            f"DEXs:  {idea['dex_a']} ↔ {idea['dex_b']}\n"
            f"Token: {idea['address']}\n"
            + (f"Why:   {reason}\n" if reason else "")
            + verify_note
            + f"\n\n⚠️ Verify the contract on {explorer} before approving. "
            "Even after approval, the scanner's live quote-verification "
            "still gates every trade.\n\n"
            "Reply /approve to add it to the LIVE scan rotation, or /reject.",
            chat_id,
        )

    # v17.27 — Solana counterpart to _cmd_ai_idea() above, same shape and
    # same safety contract: schema-constrained proposal, grounded against
    # live chain state, nothing watched until an explicit /approve. See
    # qwen_client.py's propose_solana_route() and solana_arb.py's
    # verify_solana_mint() for the two pieces this wires together.
    async def _cmd_ai_idea_sol(self, chat_id: int) -> None:
        if self._qwen is None or not getattr(self._qwen, "enabled", False):
            await self._safe_send(
                "🤖 AI is not configured — set QWEN_API_KEY in the Space "
                "secrets and restart.", chat_id,
            )
            return
        if self._solana is None:
            await self._safe_send(
                "🤖 The Solana leg isn't running (SOLANA_SCAN_ENABLED is off "
                "or it hasn't started yet) — nothing to add a route to.",
                chat_id,
            )
            return
        await self._safe_send("🤖 Asking the AI for one concrete Solana route idea…", chat_id)
        watching = [f"{r['base']}->{r['mid']}" for r in (self._solana.routes or [])]
        idea: dict | None = None
        reject_reasons: list[str] = []
        for _attempt in range(self._AI_IDEA_MAX_ATTEMPTS):
            candidate = await self._qwen.propose_solana_route(already_watching=watching)
            if not candidate:
                reject_reasons.append(
                    getattr(self._qwen, "last_error", None)
                    or "model returned no usable suggestion"
                )
                continue
            base_v = await self._solana.verify_solana_mint(candidate["base"])
            mid_v = await self._solana.verify_solana_mint(candidate["mid"])
            if not base_v.get("exists") or not mid_v.get("exists"):
                bad = base_v if not base_v.get("exists") else mid_v
                reject_reasons.append(
                    f"{candidate['base']}->{candidate['mid']} — {bad.get('reason', 'not verified')}"
                )
                continue
            idea = candidate
            break
        if not idea:
            detail = reject_reasons[-1] if reject_reasons else "no reason given"
            await self._safe_send(
                f"🤖 No usable suggestion after {self._AI_IDEA_MAX_ATTEMPTS} "
                f"tries ({detail}). Try /ai_idea_sol again.",
                chat_id,
            )
            return
        reason = str(idea.pop("reason", "") or "")
        idea["chain"] = "SOL"
        self._pending_ai_pair = idea
        await self._safe_send(
            "🤖 AI Solana route idea — awaiting YOUR approval\n\n"
            f"Route: {idea['base']}->{idea['mid']} (via Jupiter)\n"
            + (f"Why:   {reason}\n" if reason else "")
            + "✅ Both mints verified on-chain.\n\n"
            "⚠️ Even after approval, the same live re-quote/margin/"
            "simulation gates decide whether it ever trades.\n\n"
            "Reply /approve to add it to the LIVE scan rotation, or /reject.",
            chat_id,
        )

    async def _cmd_ai_do(self, chat_id: int, args: list[str]) -> None:
        if self._qwen is None or not getattr(self._qwen, "enabled", False):
            await self._safe_send(
                "🤖 AI is not configured — set QWEN_API_KEY in the Space "
                "secrets and restart.", chat_id,
            )
            return
        # v17.27 ("fix his mind ... make him small agent") — a leading "sol"
        # token targets the Solana leg instead of ARB, mirroring
        # /ai_idea_sol's own separate command rather than trying to guess
        # the chain from free-form wording. Plain "/ai_do ..." is
        # byte-for-byte unchanged (still ARB, same as always).
        target_sol = bool(args) and args[0].strip().lower() == "sol"
        request = " ".join(args[1:] if target_sol else args).strip()
        if not request:
            await self._safe_send(
                "Usage: /ai_do <what you want her to do>\n"
                "       /ai_do sol <what you want her to do>\n\nExamples:\n"
                "/ai_do set loan to 20000\n"
                "/ai_do set min profit to 5\n"
                "/ai_do pause hunting\n"
                "/ai_do resume\n"
                "/ai_do stop watching BNB/USDC\n"
                "/ai_do sol set min profit to 5\n"
                "/ai_do sol pause hunting\n"
                "/ai_do sol stop watching SOL/BONK\n\n"
                "She proposes ONE action; nothing changes until you /approve.",
                chat_id,
            )
            return
        await self._safe_send("🤖 Working out one action…", chat_id)
        if target_sol:
            if self._solana is None:
                await self._safe_send(
                    "🤖 The Solana leg isn't running (SOLANA_SCAN_ENABLED is "
                    "off or it hasn't started yet).", chat_id,
                )
                return
            sol = self._solana
            opps = sol.last_opportunities or []
            best = max(opps, key=lambda o: o.net_profit_usd) if opps else None
            context = (
                f"routes={len(sol.routes)}; "
                f"min_profit=${float(sol.min_profit):.2f}; "
                f"paused={getattr(sol, 'paused', False)}; "
                f"last={best.near_miss_line(float(sol.min_profit)) if best else 'no data yet'}"
            )
            action = await self._qwen.propose_action(request, context=context, chain="SOL")
        else:
            last = _recent_results[-1] if _recent_results else {}
            try:
                pair_count = len(getattr(self._scanner, "_pairs", []) or [])
            except Exception:
                pair_count = 0
            context = (
                f"pairs={pair_count}; "
                f"loan=${getattr(self._scanner, 'loan_amount', 0):,.0f}; "
                f"min_profit=${getattr(self._scanner, 'min_profit', 0):,.2f}; "
                f"paused={getattr(self._scanner, 'paused', False)}; "
                f"last={last.get('signal', '?')} "
                f"net ${float(last.get('netAfterFee', 0) or 0):.2f}"
            )
            # v1.8 FIX — same wrong-chain default as /ai_idea above; an
            # approved add_pair action must target the live chain (ARB).
            action = await self._qwen.propose_action(request, context=context, chain="ARB")
        if not action:
            await self._safe_send(
                "🤖 Couldn't turn that into a valid action — try rephrasing, "
                "e.g. '/ai_do set loan to 20000'.", chat_id,
            )
            return
        if action.get("action") == "none":
            reason = action.get("reason") or "not a supported control action"
            await self._safe_send(
                f"🤖 I can't do that as an action ({reason}).\n"
                "For questions use /ai; to add a pair use /ai_idea "
                "(or /ai_idea_sol for Solana).", chat_id,
            )
            return
        self._pending_ai_action = action
        await self._safe_send(
            "🤖 AI action — awaiting YOUR approval\n\n"
            f"Proposed: {action.get('desc', '(action)')}\n\n"
            "Reply /approve to apply it, or /reject.\n"
            "⚠️ Every trade still clears the same live on-chain gates "
            "regardless of this setting.",
            chat_id,
        )

    def _apply_ai_action(self, action: dict) -> tuple[bool, str]:
        """Execute ONE operator-approved AI action against the live scanner
        (or, for chain="SOL" actions, the live SolanaArbEngine — v17.27).
        Never raises. Setting changes are runtime-only (they revert on
        restart — set the matching env var for a permanent change); added
        pairs/routes persist via ai_pairs.json/ai_solana_routes.json exactly
        like /ai_idea → /approve / /ai_idea_sol → /approve."""
        try:
            kind = action.get("action")
            # v17.27 — pause/resume/set_min_profit are now proposed WITH a
            # "chain" tag (see qwen_client.py's matching v1.22 entry); route
            # to the Solana engine instead of the ARB Scanner when it says
            # SOL. add_route/remove_route only ever exist for chain="SOL".
            if kind in ("pause", "resume", "set_min_profit") and str(action.get("chain", "")).upper() == "SOL":
                if self._solana is None:
                    return False, "the Solana leg isn't running anymore"
                sol = self._solana
                if kind == "pause":
                    sol.paused = True
                    return True, ("Solana hunting PAUSED — nothing trades until you "
                                  "resume (/ai_do sol resume → /approve). Runtime-only.")
                if kind == "resume":
                    sol.paused = False
                    return True, "Solana hunting RESUMED."
                val = float(action["value"])
                sol.min_profit = val
                return True, (f"Solana minimum profit floor set to ${val:,.2f} for "
                              "this run — set SOLANA_MIN_PROFIT_USD to persist.")
            if kind == "add_route":
                if self._solana is None:
                    return False, "the Solana leg isn't running anymore"
                ok, msg = self._solana.add_route(action["base"], action["mid"])
                if ok:
                    self._persist_ai_solana_route(action)
                return ok, msg
            if kind == "remove_route":
                if self._solana is None:
                    return False, "the Solana leg isn't running anymore"
                return self._solana.remove_route(action["base"], action["mid"])

            sc = self._scanner
            if kind == "pause":
                sc.paused = True
                return True, ("hunting PAUSED — nothing trades until you resume "
                              "(/ai_do resume → /approve). Runtime-only.")
            if kind == "resume":
                sc.paused = False
                return True, "hunting RESUMED."
            if kind == "set_loan":
                val = float(action["value"])
                sc.loan_amount = val
                return True, (f"flash-loan size set to ${val:,.0f} for this run — "
                              "set FLASH_LOAN_USD in Space Variables to persist.")
            if kind == "set_min_profit":
                val = float(action["value"])
                sc.min_profit = val
                return True, (f"minimum profit floor set to ${val:,.2f} for this run — "
                              "set MIN_PROFIT_USD to persist.")
            if kind == "remove_pair":
                return sc.remove_pair(
                    action["base"], action["stable"], action.get("chain", "BSC"),
                )
            if kind == "add_pair":
                pair = dict(action["pair"])
                ok, msg = sc.add_pair(pair)
                if ok:
                    self._persist_ai_pair(pair)
                return ok, msg
            return False, f"unknown action {kind!r}"
        except Exception as exc:
            return False, f"action failed: {exc}"

    async def _cmd_approve(self, chat_id: int) -> None:
        # v18.36 — a pending AI ACTION takes priority over a pending pair.
        if self._pending_ai_action is not None:
            action = self._pending_ai_action
            self._pending_ai_action = None
            ok, msg = self._apply_ai_action(action)
            await self._safe_send(
                (f"✅ Approved — {msg}" if ok else f"❌ Not applied: {msg}"),
                chat_id,
            )
            return
        idea = self._pending_ai_pair
        if not idea:
            await self._safe_send(
                "Nothing pending — get a suggestion with /ai_idea, /ai_idea_sol, "
                "or /ai_do first.",
                chat_id,
            )
            return
        self._pending_ai_pair = None
        # v17.27 — a pending pair from /ai_idea_sol is tagged chain="SOL" and
        # has a completely different shape (base/mid, no stable/address/
        # dex_a/dex_b) — route it to SolanaArbEngine.add_route() instead of
        # the ARB/BSC Scanner.add_pair() path below.
        if str(idea.get("chain", "")).upper() == "SOL":
            if self._solana is None:
                await self._safe_send(
                    "❌ Not added: the Solana leg isn't running anymore.", chat_id,
                )
                return
            ok, msg = self._solana.add_route(idea["base"], idea["mid"])
            if not ok:
                await self._safe_send(f"❌ Not added: {msg}", chat_id)
                return
            self._persist_ai_solana_route(idea)
            await self._safe_send(
                f"✅ Approved — {msg}.\n"
                "It joins the Solana scan rotation next cycle and persists "
                "across restarts (ai_solana_routes.json). The same live "
                "re-quote/margin/simulation gates still decide whether it "
                "ever trades.",
                chat_id,
            )
            return
        ok, msg = self._scanner.add_pair(dict(idea))
        if not ok:
            await self._safe_send(f"❌ Not added: {msg}", chat_id)
            return
        self._persist_ai_pair(idea)
        await self._safe_send(
            f"✅ Approved — {msg}.\n"
            "It joins the scan rotation next cycle and persists across "
            "restarts (ai_pairs.json). The same live quote-verification "
            "gates still decide whether it ever trades.",
            chat_id,
        )

    async def _cmd_reject(self, chat_id: int) -> None:
        # v18.36 — reject a pending AI action first, if any.
        if self._pending_ai_action is not None:
            act = self._pending_ai_action
            self._pending_ai_action = None
            await self._safe_send(
                f"🗑️ Rejected action — {act.get('desc', '(action)')}. Nothing changed.",
                chat_id,
            )
            return
        if self._pending_ai_pair is None:
            await self._safe_send("Nothing pending to reject.", chat_id)
            return
        rejected = self._pending_ai_pair
        self._pending_ai_pair = None
        # v17.27 — a Solana route has no 'stable' key; show base->mid instead
        # of the ARB-shaped "base/stable" for a chain="SOL" rejection.
        if str(rejected.get("chain", "")).upper() == "SOL":
            await self._safe_send(
                f"🗑️ Rejected {rejected.get('base', '?')}->"
                f"{rejected.get('mid', '?')} — discarded.",
                chat_id,
            )
            return
        await self._safe_send(
            f"🗑️ Rejected {rejected.get('base', '?')}/"
            f"{rejected.get('stable', '?')} — discarded.",
            chat_id,
        )

    def _persist_ai_pair(self, pair: dict) -> None:
        """Append an approved pair to ai_pairs.json — best-effort: a failed
        write only costs restart-persistence, never the running scan."""
        if not self._ai_pairs_path:
            return
        try:
            from pathlib import Path as _Path
            path = _Path(self._ai_pairs_path)
            existing: list = []
            if path.exists():
                loaded = json.loads(path.read_text() or "[]")
                if isinstance(loaded, list):
                    existing = loaded
            existing.append(pair)
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(json.dumps(existing, indent=2))
            logger.info("[CommandHandlers] AI pair persisted to %s", path)
        except Exception as exc:
            logger.warning(
                "[CommandHandlers] could not persist AI pair (%s) — it stays "
                "active until restart.", exc,
            )

    def _persist_ai_solana_route(self, route: dict) -> None:
        """v17.27 — Solana counterpart to _persist_ai_pair(), same best-effort
        contract (a failed write only costs restart-persistence, never the
        running scan). Separate file (ai_solana_routes.json) because the
        schema is different (base/mid, no stable/address/dex_a/dex_b) and
        bot.py loads/merges the two independently at startup."""
        if not self._ai_solana_routes_path:
            return
        try:
            from pathlib import Path as _Path
            path = _Path(self._ai_solana_routes_path)
            existing: list = []
            if path.exists():
                loaded = json.loads(path.read_text() or "[]")
                if isinstance(loaded, list):
                    existing = loaded
            existing.append({"base": route["base"], "mid": route["mid"]})
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(json.dumps(existing, indent=2))
            logger.info("[CommandHandlers] AI Solana route persisted to %s", path)
        except Exception as exc:
            logger.warning(
                "[CommandHandlers] could not persist AI Solana route (%s) — "
                "it stays active until restart.", exc,
            )

    # ── Safe send (never raises) ──────────────────────────────────────────────

    async def _safe_send(self, text: str, chat_id: int) -> None:
        try:
            await self._tg.send_message(text, chat_id=chat_id, parse_mode=None)
        except Exception as exc:
            logger.error("_safe_send failed for chat %s: %s", chat_id, exc)