gateway_controller.py

#!/usr/bin/env python3
"""
IB Gateway Controller — Python + in-JVM Java agent replacement for IBC.

Single-file controller that:
  1. Launches IB Gateway directly via its install4j launcher.
  2. Drives the login dialog, 2FA dialog, and configuration dialogs via
     an in-JVM Java agent (loaded with -javaagent:) that exposes Swing
     operations over a Unix domain socket.
  3. Verifies every action by reading state back through the same agent.
  4. Signals readiness via /tmp/gateway_ready so socat starts only AFTER
     login succeeds and the main window is up.
  5. Stays alive monitoring the JVM and watching for re-auth events.

Why an in-JVM agent and not xdotool: xdotool synthesizes X events but
Swing's AWT subsystem filters them. AT-SPI2 (the original approach) was
abandoned in v0.5.12 after the java-atk-wrapper bridge was found to
deadlock on JProgressBar.setValue calls during login. The agent uses
Swing's own JTextField.setText() / AbstractButton.doClick() and reads
state back via the same APIs — the only mechanisms that actually work
against Gateway's hardened login form.

Why a single Python file: the upstream maintainer (gnzsnz) doesn't write
Java or Rust, and wants something he can read, debug, and patch when IB
ships a new dialog on a Friday night. Python stdlib + a small in-JVM
agent hits that bar.

Reference material: Lcstyle's ibctl (https://github.com/Lcstyle/ibctl)
for dialog catalog and edge cases. Original work by @rlktradewright (IBC).
"""

import base64
import binascii
import enum
import hashlib
import hmac
import json
import logging
import os
import re
import signal
import socket
import struct
import subprocess
import sys
import threading
import time
from datetime import datetime, time as dtime, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from zoneinfo import ZoneInfo


__version__ = "0.16.3"

# Wall-clock timestamp recorded when the controller module loads. Reported
# by the /health endpoint as `uptime_seconds` so monitoring can spot a
# container that just restarted vs. one that's been stable.
_CONTROLLER_START_TS = time.time()


# ── Controller state machine ──────────────────────────────────────────
#
# The controller proceeds through these states in order. Each state
# transition is logged so the current position is visible in the output.
# A RESTART command resets to LAUNCHING and re-drives the sequence.
#
#   INIT → LAUNCHING → AGENT_WAIT → APP_DISCOVERY → LOGIN → POST_LOGIN
#   → TWO_FA → DISCLAIMERS → API_WAIT → CONFIG → COMMAND_SERVER → READY
#   → MONITORING
#          ↑                                              │
#          └──── (RESTART / re-auth on session loss) ─────┘
#
# This is not a formal state machine library (gnzsnz suggested
# python-statemachine — open to refactoring). But the states are
# explicit, logged at each transition, and visible in the output so
# the flow is clear to anyone reading the logs or the code.

class State(enum.Enum):
    INIT = "INIT"
    LAUNCHING = "LAUNCHING"
    AGENT_WAIT = "AGENT_WAIT"
    APP_DISCOVERY = "APP_DISCOVERY"
    LOGIN = "LOGIN"
    POST_LOGIN = "POST_LOGIN"
    TWO_FA = "TWO_FA"
    DISCLAIMERS = "DISCLAIMERS"
    API_WAIT = "API_WAIT"
    CONFIG = "CONFIG"
    COMMAND_SERVER = "COMMAND_SERVER"
    READY = "READY"
    MONITORING = "MONITORING"
    HALTED = "HALTED"
    # Gateway is up but reports no connection to IBKR, and recovery is in
    # progress (upstream watchdog, field report 2026-10-03).
    UPSTREAM_DOWN = "UPSTREAM_DOWN"


_current_state = State.INIT


def _set_state(new_state):
    """Transition to a new controller state. Logs the transition."""
    global _current_state
    old = _current_state
    _current_state = new_state
    log.info(f"[STATE: {new_state.value}]")


# ── Config from environment ─────────────────────────────────────────────
# Same env var names IBC uses, so existing docker-compose files keep working.

USERNAME = os.environ.get("TWS_USERID", "")
PASSWORD = os.environ.get("TWS_PASSWORD", "")
TRADING_MODE = os.environ.get("TRADING_MODE", "paper").lower()
TOTP_SECRET = os.environ.get("TWOFACTOR_CODE", "")

# When TRADING_MODE=paper, prefer the *_PAPER credentials if set. This
# mirrors run.sh's dual-mode logic, where the paper instance is started
# with TWS_USERID = $TWS_USERID_PAPER. Users running TRADING_MODE=both
# in production typically set TWS_USERID to their LIVE account and
# TWS_USERID_PAPER to their paper account; our paper-only test mode
# should pick the paper account, otherwise IBKR rejects the login as
# 'multiple paper trading users associated with this user'.
if TRADING_MODE == "paper":
    _paper_user = os.environ.get("TWS_USERID_PAPER", "")
    _paper_pass = os.environ.get("TWS_PASSWORD_PAPER", "")
    if _paper_user:
        USERNAME = _paper_user
    if _paper_pass:
        PASSWORD = _paper_pass

# IBKR regional server hostname. IB accounts are bound to one regional
# data center (ndc1.ibllc.com / cdc1.ibllc.com / zdc1.ibllc.com / etc.).
# Without this set, Gateway tries the default and fails the SSL
# handshake on the misc URLs port for accounts hosted elsewhere.
# Users can find their server in their IB account portal, or by running
# Gateway interactively (via VNC) once and checking the Peer line in
# the resulting Jts/jts.ini.
#
# Same _PAPER convention: an IB user can have their LIVE account on one
# server and PAPER on another (we observed this in real-world testing).
def _validate_hostname(value, varname):
    """Strict hostname validation for env vars we write into jts.ini.

    IBKR regional servers are always simple hostnames like
    `cdc1.ibllc.com`, so anything with whitespace, newlines, or
    control characters is either a typo or an injection attempt.
    Defends against the theoretical attack of a user setting e.g.
    `TWS_SERVER="cdc1.ibllc.com\\n[Logon]\\nEvil=yes"` which would
    otherwise inject extra .ini sections when we write jts.ini.
    """
    if not value:
        return value
    import re as _re
    if not _re.fullmatch(r"[A-Za-z0-9._-]+", value):
        raise ValueError(
            f"{varname}={value!r} is not a valid hostname "
            "(expected DNS label characters only). Refusing to proceed "
            "because this value would be written into jts.ini and a "
            "malformed value could inject unintended .ini content."
        )
    return value


_IPV4_RE = re.compile(r"(?<![\w.])(?:\d{1,3}\.){3}\d{1,3}(?!\w|\.\d)")
_IPV6_RE = re.compile(r"(?<![\w:])(?:[0-9A-Fa-f]{0,4}:){2,7}[0-9A-Fa-f]{0,4}(?![\w:])")


def _mask_public_ips(s):
    """Replace public IP addresses with [IP]. Gateway's dialogs and logs name
    the address of a competing session or a connecting client; private,
    loopback and link-local addresses (Docker's own) stay, as they're what
    an operator needs for Trusted IPs and identify no one."""
    import ipaddress as _ip

    def repl(m):
        try:
            a = _ip.ip_address(m.group(0))
        except ValueError:
            return m.group(0)
        keep = a.is_private or a.is_loopback or a.is_link_local or a.is_unspecified
        return m.group(0) if keep else "[IP]"
    return _IPV6_RE.sub(repl, _IPV4_RE.sub(repl, s))


def _redact_logs(s):
    """Strip sensitive patterns from a log line before we emit it.

    Gateway window titles like "DU9999999 Trader Workstation
    Configuration (Simulated Trading)" include the user's account
    number (DU prefix for paper, U prefix for live). When
    CONTROLLER_DEBUG=1 dumps modal windows, those titles appear in
    logs users may share to ask for help. Redact them so the default
    debug-log experience doesn't leak identifiers.

    Also masks the configured IBKR usernames (TWS_USERID and
    TWS_USERID_PAPER) wherever they appear, and public IP addresses: the
    upstream watchdog logs dialog text and launcher.log lines verbatim
    before a relaunch, and the existing-session dialog names the other
    session's address.
    """
    if not isinstance(s, str):
        return s
    import re as _re
    # Paper accounts start with DU, live with U + digits (IBKR convention)
    # (advisor and broker accounts use F/I, and DF/DI on paper).
    s = _re.sub(r"\b(DU|DF|DI|U|F|I)\d{5,10}\b", r"\1[REDACTED]", s)
    s = _mask_public_ips(s)
    for name in {os.environ.get("TWS_USERID", ""),
                 os.environ.get("TWS_USERID_PAPER", "")}:
        name = name.strip()
        if len(name) >= 3:
            s = _re.sub(_re.escape(name), "[USER]", s, flags=_re.IGNORECASE)
    return s


TWS_SERVER = _validate_hostname(os.environ.get("TWS_SERVER", ""), "TWS_SERVER")
if TRADING_MODE == "paper":
    _paper_server = _validate_hostname(
        os.environ.get("TWS_SERVER_PAPER", ""), "TWS_SERVER_PAPER")
    if _paper_server:
        TWS_SERVER = _paper_server

# Gateway install path (where the install4j launcher + JRE live). This
# is shared across dual-mode instances — both JVMs run the same Gateway
# binary.
TWS_PATH = os.environ.get("TWS_PATH", os.path.expanduser("~/Jts"))
TWS_VERSION = os.environ.get("TWS_MAJOR_VRSN", "")

# Per-instance config/state directory. In dual-mode containers, run.sh
# sets TWS_SETTINGS_PATH to Jts_live / Jts_paper so each Gateway JVM
# has isolated state (its own jts.ini, encrypted state dir, autorestart
# tokens, launcher.log). Falls back to TWS_PATH for single-mode.
#
# This is what we pass to Gateway as -DjtsConfigDir and what we write
# jts.ini / read warm state from. TWS_PATH is used only for locating
# the install4j launcher.
JTS_CONFIG_DIR = os.environ.get("TWS_SETTINGS_PATH") or TWS_PATH

# Readiness signal file. Each dual-mode instance uses a distinct path so
# run.sh can wait for each instance independently (live vs paper).
READY_FILE = os.environ.get("CONTROLLER_READY_FILE", "/tmp/gateway_ready")

# In-JVM input agent — provides text input that's structurally impossible
# from outside the JVM. The agent jar is loaded via -javaagent: in
# INSTALL4J_ADD_VM_PARAMS and listens on this Unix socket.
AGENT_JAR = os.environ.get("GATEWAY_INPUT_AGENT_JAR",
                           os.path.expanduser("~/gateway-input-agent.jar"))
AGENT_SOCKET = os.environ.get("GATEWAY_INPUT_AGENT_SOCKET",
                              "/tmp/gateway-input.sock")

# Optional warm-state directory. If set and exists, the controller copies
# its contents into TWS_PATH before launching Gateway. Used to seed a
# fresh container with previously-saved settings (jts.ini, encrypted
# account state, autorestart tokens) so Gateway connects to the correct
# regional server (e.g. cdc1.ibllc.com vs the default ndc1.ibllc.com)
# and can bypass full re-auth via the autorestart token if it's recent.
WARM_STATE_DIR = os.environ.get("GATEWAY_WARM_STATE", "")

# Spike-mode flag — when set, the controller will exit cleanly after
# clicking Log In instead of waiting for a real main window. Used by the
# spike test harness with bogus credentials.
TEST_MODE = os.environ.get("CONTROLLER_TEST_MODE", "") == "1"

# Our Gateway JVM's OS process ID. Populated in main() right after
# agent_wait_ready() succeeds, from the agent's GET_PID response. Used
# by find_app() to pick "this controller's" Gateway instance out of the
# AT-SPI desktop tree in dual-mode containers where two 'IBKR Gateway'
# apps are present simultaneously. Stays None until the agent is up.
JVM_PID = None

# Absolute path of the install4j launcher script the controller spawned
# (set by launch_gateway). Used to find install4j's
# ``.install4j/restarter.log`` next to it when deciding whether a JVM
# exit was Gateway's own daily auto-restart (issue #23).
_GATEWAY_LAUNCHER_PATH = None

# mtime of the install4j restarter.log that the last adoption attempt
# acted on. A restarter.log stays "fresh" for 120s, so without this an
# adopted JVM that dies inside that window would send the controller
# through a second pointless adoption wait for the same restart.
_LAST_ADOPTED_RESTART_MTIME = None

# The Gateway JVM subprocess and its AT-SPI application accessible.
# Made module-global so the Phase 2.4 command server's RESTART handler
# can tear down the current JVM and re-launch + re-login in place,
# updating these globals so the monitor loop automatically picks up
# the new references without having to be restarted itself.
GATEWAY_PROC = None
CURRENT_APP = None

# Product switch: 'gateway' (default) or 'tws'. Phase 2.3 lets the same
# controller drive IB Gateway OR Trader Workstation from the same image.
# TWS exposes a different AT-SPI application name than Gateway, so
# find_app() scans both names and a couple of common variants.
GATEWAY_OR_TWS = os.environ.get("GATEWAY_OR_TWS", "gateway").strip().lower()
if GATEWAY_OR_TWS == "tws":
    APP_NAME_CANDIDATES = ["Trader Workstation", "IB Trader Workstation", "TWS"]
else:
    APP_NAME_CANDIDATES = ["IBKR Gateway"]


# ── Logging ─────────────────────────────────────────────────────────────
#
# v0.6.1: include TRADING_MODE as a fixed prefix on every log line. In
# dual mode (TRADING_MODE=both, which run.sh splits into two parallel
# controller processes — one "live", one "paper") both controllers'
# stdout interleaves into the same `docker logs <container>` stream
# with no way to tell their lines apart. That made bug reports like
# 2026-05-01's "post-login config skipped for live but ran for paper"
# impossible to diagnose from logs alone (pre-v0.6.1 the reporter had
# to pair `[STATE: CONFIG]` lines with subsequent context to guess
# which controller emitted them). The mode prefix is fixed at module
# load — TRADING_MODE doesn't change for the life of a controller
# process, so f-string interpolation here is correct.

logging.basicConfig(
    level=logging.DEBUG if os.environ.get("CONTROLLER_DEBUG") else logging.INFO,
    format=f"%(asctime)s [%(levelname)s] [{TRADING_MODE}] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("controller")


# ── TOTP generation (stdlib only — no oathtool, no pyotp) ───────────────

def _validate_totp_secret(secret):
    """Return None if ``secret`` is usable by ``generate_totp``, else a
    one-line explanation for the operator.

    Empty is fine: it selects IB Key push mode. Anything else must be
    the base32 *secret* IBKR shows when enrolling Mobile Authenticator.
    The mistake this exists to catch (issue #7, 2026-09-11): pasting a
    six-digit *generated* code instead. ``base64.b32decode`` rejects
    that, and before this check the failure was an uncaught
    ``binascii.Error`` traceback the moment the 2FA dialog appeared.
    """
    if not secret:
        return None
    if secret.strip().isdigit() and len(secret.strip()) in (6, 7, 8):
        return (f"TWOFACTOR_CODE looks like a generated {len(secret.strip())}-digit "
                "code. It must be the base32 SECRET from IBKR's Mobile "
                "Authenticator enrolment (a long string of letters and the "
                "digits 2-7); the controller generates the current code from it.")
    try:
        base64.b32decode(secret, casefold=True)
        return None
    except (binascii.Error, ValueError) as e:
        hint = ""
        if " " in secret:
            try:
                base64.b32decode(secret.replace(" ", ""), casefold=True)
                hint = " Removing the spaces would make it valid."
            except (binascii.Error, ValueError):
                pass
        return (f"TWOFACTOR_CODE is not valid base32 ({e}). It must be the "
                "base32 secret from IBKR's Mobile Authenticator enrolment, "
                f"not a generated code.{hint}")


def generate_totp(secret_b32, period=30, digits=6):
    """Generate a TOTP code from a base32 secret. Stdlib only."""
    key = base64.b32decode(secret_b32, casefold=True)
    counter = struct.pack(">Q", int(time.time()) // period)
    mac = hmac.new(key, counter, hashlib.sha1).digest()
    offset = mac[-1] & 0x0F
    code = struct.unpack(">I", mac[offset:offset + 4])[0] & 0x7FFFFFFF
    return str(code % (10 ** digits)).zfill(digits)


# ── AT-SPI helpers ──────────────────────────────────────────────────────

def safe(fn, default=None):
    """Run an AT-SPI call with retry. AT-SPI calls can timeout under load."""
    for _ in range(3):
        try:
            return fn()
        except Exception:
            time.sleep(0.2)
    return default


class _AppHandle:
    """Lightweight handle representing the controller's identified Gateway JVM.

    The class predates v0.5.12 — it used to expose a pyatspi-Accessible-like
    surface (get_role_name/get_state_set/get_child_count/...) so callers
    could walk the AT-SPI tree off it. v0.5.12 disabled the AT-SPI bridge
    in the JVM, and v0.5.14 removed the dead tree-walking helpers
    (find_descendant / wait_for / get_states / _read_text / click(node) /
    set_text(node, ...) / _dump_tree). The handle now carries only the
    name + PID — what surviving callers actually use, plus a passthrough
    argument for legacy signatures (handle_login(app), attempt_reauth(app),
    do_restart_in_place(app), monitor_loop(app)).
    """

    def __init__(self, name="IBKR Gateway", pid=None):
        self._name = name
        self._pid = pid

    def get_name(self):
        return self._name

    def get_process_id(self):
        return self._pid


def find_app(name_substring, timeout=120, match_pid=None):
    """Return an app handle carrying the JVM PID reported by the agent.

    Pre-v0.5.12 this polled the AT-SPI desktop tree until an entry
    matching ``name_substring`` appeared. v0.5.12 disabled the AT-SPI
    bridge in the JVM (``launch_gateway`` passes
    ``-Djavax.accessibility.assistive_technologies=``), so the desktop
    tree never populates — polling would always time out. We
    short-circuit to an ``_AppHandle`` carrying the agent-reported JVM
    PID so existing callers (which use the return value only for
    logging and as a passthrough argument to handle_login /
    attempt_inplace_relogin / etc.) keep working without every callsite
    having to know that AT-SPI is gone.

    Returns the handle immediately. The ``timeout`` argument is
    ignored. Returns ``None`` only if the agent's ``GET_PID`` never
    succeeded (``agent_get_pid`` returned ``None`` at controller
    startup), since in that case dual-mode containers can't safely
    identify "their own" JVM and the caller should treat that as a
    genuine launch failure.
    """
    pid = match_pid if match_pid is not None else JVM_PID
    if pid is None:
        # Agent didn't report a PID — refuse to claim discovery succeeded.
        # Caller treats None as fatal in main() / do_restart_in_place.
        return None
    name = name_substring if isinstance(name_substring, str) else (
        name_substring[0] if name_substring else "IBKR Gateway")
    return _AppHandle(name=name, pid=pid)


# ── In-JVM agent client ─────────────────────────────────────────────────

def _agent_request(line, timeout=10.0):
    """Send a single line to the input agent and read its single-line response.
    Opens a fresh Unix socket connection per request — the agent is
    single-threaded and the controller is its only client, so connection
    pooling buys us nothing."""
    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    s.settimeout(timeout)
    try:
        s.connect(AGENT_SOCKET)
        s.sendall((line + "\n").encode("utf-8"))
        # Read until newline
        buf = b""
        while b"\n" not in buf:
            chunk = s.recv(4096)
            if not chunk:
                break
            buf += chunk
        return buf.split(b"\n", 1)[0].decode("utf-8", errors="replace")
    finally:
        s.close()


def agent_wait_ready(timeout=60):
    """Wait until the agent's Unix socket exists and answers PING."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if os.path.exists(AGENT_SOCKET):
            try:
                resp = _agent_request("PING", timeout=2)
                if resp.startswith("OK"):
                    return True
            except Exception:
                pass
        time.sleep(0.3)
    return False


def agent_get_pid():
    """Return the JVM's OS process ID via the agent, or None on failure.

    Used to disambiguate "this controller's Gateway JVM" from any other
    Gateway JVM running in the same container (dual-mode case: both live
    and paper JVMs appear as 'IBKR Gateway' in the AT-SPI desktop tree).
    """
    try:
        resp = _agent_request("GET_PID", timeout=2)
    except Exception as e:
        log.warning(f"agent GET_PID failed: {type(e).__name__}: {e}")
        return None
    if not resp.startswith("OK "):
        log.warning(f"agent GET_PID unexpected response: {resp!r}")
        return None
    try:
        return int(resp[3:].strip())
    except ValueError:
        log.warning(f"agent GET_PID non-integer response: {resp!r}")
        return None


def agent_settext(name, text):
    """Set text on a Swing JTextComponent by accessible name. Returns True on success."""
    try:
        resp = _agent_request(f"SETTEXT {name} {text}")
    except Exception as e:
        log.error(f"agent SETTEXT {name!r}: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent SETTEXT {name!r}: {resp}")
    return False


def agent_gettext(name):
    """Read text from a Swing JTextComponent by accessible name. Returns string or None."""
    try:
        resp = _agent_request(f"GETTEXT {name}")
    except Exception as e:
        log.error(f"agent GETTEXT {name!r}: {type(e).__name__}: {e}")
        return None
    if resp.startswith("OK "):
        return resp[3:]
    if resp == "OK":
        return ""
    log.error(f"agent GETTEXT {name!r}: {resp}")
    return None


def _login_button_labels(mode=None):
    """Return ``(expected, fallback)`` Log In button labels for a trading
    mode. Gateway renames the button: "Log In" on live, "Paper Log In"
    on paper. Probing the expected one first keeps the routine case off
    the ERROR log (see the call site in ``handle_login``)."""
    if (mode if mode is not None else TRADING_MODE) == "paper":
        return ("Paper Log In", "Log In")
    return ("Log In", "Paper Log In")


def agent_click(name, quiet=False):
    """Click an AbstractButton by accessible name. Returns True on success.

    ``quiet=True`` demotes a failed click to DEBUG. Use it only where a
    miss is an expected outcome the caller handles — probing a button
    whose label varies, for instance — so that real failures stay at
    ERROR and log-scraping alert rules don't fire on routine probes.
    """
    level = log.debug if quiet else log.error
    try:
        resp = _agent_request(f"CLICK {name}")
    except Exception as e:
        level(f"agent CLICK {name!r}: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    level(f"agent CLICK {name!r}: {resp}")
    return False


# The agent waits up to 10 s for the login box to be enabled before it
# answers; the socket timeout has to outlast that wait, or the controller
# gives up just before the agent replies.
_SETTEXT_LOGIN_TIMEOUT = 15.0


def agent_settext_login_user(text):
    """Set the Gateway login frame's username via in-JVM role-based lookup.

    v0.4.2: bypasses the name-based SETTEXT path. After a failed login
    attempt, the username field can become a JComboBox autocomplete
    editor whose JTextField child has null AccessibleName, so SETTEXT
    by name ("Username") returns not_found on re-drive from
    attempt_inplace_relogin. The agent-side command finds the field by
    Swing type (first editable non-password JTextComponent on the
    window containing a JPasswordField). Waits up to 10s for the field
    to become editable (disabled during Gateway's "Attempt N:
    connecting to server" retry animation).
    """
    try:
        resp = _agent_request(f"SETTEXT_LOGIN_USER {text}",
                              timeout=_SETTEXT_LOGIN_TIMEOUT)
    except Exception as e:
        log.error(f"agent SETTEXT_LOGIN_USER: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent SETTEXT_LOGIN_USER: {resp}")
    return False


def agent_settext_login_password(text):
    """Set the Gateway login frame's password via in-JVM role-based lookup.

    Symmetric to agent_settext_login_user. Password's accessible name
    is currently stable, but role-based lookup (match on JPasswordField
    Swing type) future-proofs against name drift.
    """
    try:
        resp = _agent_request(f"SETTEXT_LOGIN_PASSWORD {text}",
                              timeout=_SETTEXT_LOGIN_TIMEOUT)
    except Exception as e:
        log.error(f"agent SETTEXT_LOGIN_PASSWORD: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent SETTEXT_LOGIN_PASSWORD: {resp}")
    return False


def agent_login_state():
    """Whether Gateway's login window is waiting for credentials.

    ``"ready"``: its password box is enabled. ``"busy"``: the window is up
    but Gateway has greyed out its fields because it is logging in on its
    own. After a daily auto-restart Gateway restores the session it saved
    and logs in with it on its own, and the window shows while it does
    (issue #76). ``None``: no login window, or no usable answer.
    """
    try:
        resp = _agent_request("LOGIN_STATE", timeout=3)
    except Exception:
        return None
    if resp == "OK ready":
        return "ready"
    if resp == "OK busy":
        return "busy"
    return None


def agent_wait_login_frame(timeout_ms=120_000):
    """Block until a showing Window containing a JPasswordField exists
    AND no other modal dialog is blocking it. Returns True on success,
    False on timeout.

    v0.4.3: replaces pyatspi ``wait_for(app, "password text")`` in
    attempt_inplace_relogin. The AT-SPI tree filters the login frame's
    password-text role while Gateway's "Attempt N: connecting to
    server" modal is up, so the old 30s wait would time out before
    Gateway's internal retry self-cleared (typical ~60s). Swing's
    ``isShowing()`` remains truthful regardless of modal overlay, and
    the Java agent additionally checks that no modal dialog is
    blocking — only returns OK when the login frame is actually
    interactable.
    """
    try:
        resp = _agent_request(
            f"WAIT_LOGIN_FRAME {int(timeout_ms)}",
            timeout=int(timeout_ms / 1000) + 10,
        )
    except Exception as e:
        log.error(f"agent WAIT_LOGIN_FRAME: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent WAIT_LOGIN_FRAME: {resp}")
    return False


def agent_settext_in_window(title_substring, text):
    """Type text into the first editable JTextComponent of the first visible
    window whose title contains the substring. Used for fields that have
    no accessible name (e.g. the Second Factor Authentication TOTP input)."""
    try:
        resp = _agent_request(f"SETTEXT_IN_WIN {title_substring}|{text}")
    except Exception as e:
        log.error(f"agent SETTEXT_IN_WIN {title_substring!r}: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent SETTEXT_IN_WIN {title_substring!r}: {resp}")
    return False


def agent_click_in_window(title_substring, button_text):
    """Click a button (matched by getText() or accessible name) inside
    a window whose title contains the substring. Used for dialogs whose
    button identifiers overlap with main window buttons."""
    try:
        resp = _agent_request(f"CLICK_IN_WIN {title_substring}|{button_text}")
    except Exception as e:
        log.error(_redact_logs(
            f"agent CLICK_IN_WIN {title_substring!r}: {type(e).__name__}: {e}"))
        return False
    if resp.startswith("OK"):
        return True
    log.error(_redact_logs(f"agent CLICK_IN_WIN {title_substring!r}: {resp}"))
    return False


# Last reply from the agent's JLIST_SELECT, so callers can tell a genuine
# miss ("ERR jlist_item_not_found" / "_ambiguous" — the configured name is
# not in the account's list, which no restart will change) from a socket
# error, which a restart may well clear. Empty when the request never got a
# reply.
_last_jlist_response = ""


def agent_jlist_select(title_substring, item_text):
    """Ask the agent to select an item by text in the first JList of a
    window whose title contains the given substring.

    Used to pick the 2FA device (e.g. \"Mobile Authenticator app\") in
    the Second Factor Authentication dialog's device list on
    multi-method accounts (issue #7): Gateway pre-selects one method
    there and the controller's AT-SPI view can't drive a JList, but the
    in-JVM agent can — mirroring IBC's SecondFactorDevice handling of
    the same dialog.
    """
    global _last_jlist_response
    _last_jlist_response = ""
    try:
        resp = _agent_request(f"JLIST_SELECT {title_substring}|{item_text}")
    except Exception as e:
        log.error(f"agent JLIST_SELECT {title_substring!r}: {type(e).__name__}: {e}")
        return False
    _last_jlist_response = resp
    if resp.startswith("OK"):
        # The agent reports the entry it picked. Since issue #33 that can
        # differ from item_text in case or spacing, so log it.
        log.info(f"agent JLIST_SELECT {title_substring!r}: {resp}")
        return True
    log.error(f"agent JLIST_SELECT {title_substring!r}: {resp}")
    return False


def agent_close_window(title_substring):
    """v0.5.6: Post a WINDOW_CLOSING event to the first showing window
    whose title contains ``title_substring``. Returns True if the event
    was dispatched (i.e. the window was found and the agent accepted
    the request), False otherwise. Does NOT wait for the window to
    actually close — callers should poll ``GATEWAY_PROC`` for exit.

    Used by ``_attempt_clean_logout`` to drive the same close path a
    user would trigger by clicking the window's X button, which hits
    Gateway's registered WindowListener. Unlike SIGTERM (which triggers
    JVM shutdown hooks on a dedicated thread), this goes through the
    EDT and Gateway's UI-level close handler — which in turn does a
    clean CCP session-close before the JVM exits, freeing the IBKR
    session slot server-side rather than stranding it.

    Short agent timeout (2s): if the agent doesn't respond quickly the
    JVM is almost certainly in a state where clean logout won't work
    anyway, and we want to fall through to SIGTERM promptly.
    """
    try:
        resp = _agent_request(f"CLOSE_WIN {title_substring}", timeout=2)
    except Exception as e:
        log.warning(
            f"agent CLOSE_WIN {title_substring!r}: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.warning(f"agent CLOSE_WIN {title_substring!r}: {resp}")
    return False


def agent_jtree_select_path(title_substring, path):
    """Select a JTree node by path. Path is slash-separated, each
    component matching node.toString(). Used to drive Gateway's
    ConfigurationTree (Configure → Settings dialog) to a specific
    section like 'API/Settings' or 'Lock and Exit'."""
    try:
        resp = _agent_request(f"JTREE_SELECT_PATH {title_substring}|{path}")
    except Exception as e:
        log.error(f"agent JTREE_SELECT_PATH {path!r}: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent JTREE_SELECT_PATH {path!r}: {resp}")
    return False


def agent_jcheck(title_substring, name, desired):
    """Set a toggle-style button (JCheckBox/JRadioButton/JToggleButton)
    to the desired state. Returns True on success, False on error.
    'desired' is a Python bool."""
    state = "true" if desired else "false"
    try:
        resp = _agent_request(f"JCHECK {title_substring}|{name}|{state}")
    except Exception as e:
        log.error(f"agent JCHECK {name!r}: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent JCHECK {name!r}: {resp}")
    return False


def agent_jstate(title_substring, name):
    """Read a toggle-style button's selected state without changing it.
    Returns True/False, or None if it couldn't be read."""
    try:
        resp = _agent_request(f"JSTATE {title_substring}|{name}")
    except Exception as e:
        log.warning(f"agent JSTATE {name!r}: {type(e).__name__}: {e}")
        return None
    m = re.match(r"OK value=(true|false)\b", resp or "")
    if m:
        return m.group(1) == "true"
    log.warning(f"agent JSTATE {name!r}: {resp}")
    return None


def agent_settext_by_label(title_substring, label_text, value):
    """Set a text field's value by matching against an adjacent
    JLabel's text. Used for config fields like 'Master API client ID'
    where the JSpinner's editor has no accessible name of its own but
    sits next to a descriptive JLabel. Returns True on success."""
    try:
        resp = _agent_request(
            f"SETTEXT_BY_LABEL {title_substring}|{label_text}|{value}")
    except Exception as e:
        log.error(f"agent SETTEXT_BY_LABEL {label_text!r}: {type(e).__name__}: {e}")
        return False
    if resp.startswith("OK"):
        return True
    log.error(f"agent SETTEXT_BY_LABEL {label_text!r}: {resp}")
    return False


def _agent_multiline(command, timeout=5):
    """Send a command and read the multi-line response (terminated by 'END')."""
    try:
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.settimeout(timeout)
        s.connect(AGENT_SOCKET)
        s.sendall((command + "\n").encode("utf-8"))
        buf = b""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            chunk = s.recv(8192)
            if not chunk:
                break
            buf += chunk
            if buf.rstrip().endswith(b"END"):
                break
        s.close()
        return buf.decode("utf-8", errors="replace")
    except Exception as e:
        log.error(f"agent {command.split()[0]}: {type(e).__name__}: {e}")
        return ""


def agent_list(filter_substring=""):
    """Ask the agent for all visible text components and buttons.

    Returns a tuple (text_names, button_names) — both sets of strings.

    Used for live state detection after Gateway transitions away from
    the login frame. AT-SPI's view of the application accessible can go
    stale after a frame teardown (child_count returns -1, tree-walking
    finds nothing) but the in-JVM agent always sees the live Swing
    component tree via Window.getWindows().
    """
    raw = _agent_multiline(f"LIST {filter_substring}")
    text_names = set()
    button_names = set()
    for line in raw.splitlines():
        if line.startswith("text "):
            n = line[5:]
            if n != "(null)":
                text_names.add(n)
        elif line.startswith("button "):
            n = line[7:]
            if n != "(null)" and n != "":
                button_names.add(n)
    return text_names, button_names


def agent_windows():
    """Ask the agent for all currently-showing top-level windows.

    Returns a list of (type, title, modal) tuples. Critical for spotting
    blocking dialogs (existing-session, EULA, info popups) that have no
    text fields and only an OK button — we can't distinguish them from
    LIST output alone.
    """
    return _parse_windows(_agent_multiline("WINDOWS"))


def _parse_windows(raw):
    out = []
    for line in raw.splitlines():
        if line in ("OK", "END") or not line:
            continue
        # Format: "<type> | <title> | modal=<bool>"
        parts = [p.strip() for p in line.split("|")]
        if len(parts) >= 3:
            type_ = parts[0]
            title = parts[1]
            modal = parts[2].endswith("true")
            out.append((type_, title, modal))
    return out


def agent_labels(filter_substring=""):
    """Ask the agent for all visible JLabel text content (HTML stripped).

    Returns a list of (window_title, label_text) tuples. Used to read
    dialog message bodies — e.g. distinguishing the existing-session
    dialog from a wrong-credentials dialog when both expose only an OK
    button.
    """
    raw = _agent_multiline(f"LABELS {filter_substring}")
    out = []
    for line in raw.splitlines():
        if line in ("OK", "END") or not line:
            continue
        # Format: "[<window_title>] <label_text>"
        if line.startswith("[") and "]" in line:
            close = line.index("]")
            wtitle = line[1:close]
            text = line[close + 1:].strip()
            out.append((wtitle, text))
    return out


def agent_window(title_substring=""):
    """Dump the full component tree of windows whose title contains the
    given substring (empty = all visible windows). Returns the raw
    multi-line string. Captures text from JLabel, JTextComponent
    (including JTextArea/JEditorPane/JTextPane that LABELS misses),
    and AbstractButton."""
    return _agent_multiline(f"WINDOW {title_substring}", timeout=10)


# ── Gateway launch ──────────────────────────────────────────────────────

def find_gateway_launcher():
    """Locate the install4j ibgateway or tws launcher script. Returns
    absolute path.

    Controlled by the GATEWAY_OR_TWS env var ('gateway' default, 'tws'
    switches to TWS). For Gateway, the install path is
      $TWS_PATH/ibgateway/<version>/ibgateway
    For TWS it's
      $TWS_PATH/tws/<version>/tws
    matching the subdir / binary name to the product. Phase 2.3 adds
    this switch so the same controller drives either product from the
    same image with different env vars.
    """
    product = os.environ.get("GATEWAY_OR_TWS", "gateway").strip().lower()
    if product == "tws":
        subdir = "tws"
        binary = "tws"
    else:
        subdir = "ibgateway"
        binary = "ibgateway"

    if TWS_VERSION:
        candidate = os.path.join(TWS_PATH, subdir, TWS_VERSION, binary)
        if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
            return candidate
    # Fallback: pick the highest version directory
    root = os.path.join(TWS_PATH, subdir)
    if not os.path.isdir(root):
        return None
    for v in sorted(os.listdir(root), reverse=True):
        candidate = os.path.join(root, v, binary)
        if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
            return candidate
    return None


def apply_warm_state():
    """If GATEWAY_WARM_STATE points at a directory, copy its contents into
    JTS_CONFIG_DIR so Gateway starts with previously-saved state.

    This is the workaround for the cold-start SSL handshake failure: the
    user's account is served by a regional server (e.g. cdc1.ibllc.com)
    that's NOT the default `ndc1.ibllc.com` Gateway tries on first boot.
    The warm jts.ini has the correct `Peer=` and `SupportsSSL=` cache,
    so Gateway routes to the right server. Recent autorestart tokens in
    the warm state may also let Gateway skip the full re-auth.
    """
    # IBC-compat: TWS_COLD_RESTART=yes forces a cold restart by skipping
    # the warm state copy entirely. Users who suspect stale state is
    # causing problems can set this to force Gateway to start fresh
    # (and also to clear any existing jts.ini, because ensure_jts_ini
    # will regenerate it).
    if _coerce_yes_no(os.environ.get("TWS_COLD_RESTART", "")) is True:
        log.info("TWS_COLD_RESTART=yes — skipping warm state application")
        return

    if not WARM_STATE_DIR:
        return
    if not os.path.isdir(WARM_STATE_DIR):
        log.warning(f"GATEWAY_WARM_STATE={WARM_STATE_DIR} is not a directory")
        return
    if not os.path.isabs(WARM_STATE_DIR):
        log.warning(f"GATEWAY_WARM_STATE={WARM_STATE_DIR} is not an absolute "
                    "path — refusing to apply warm state from a relative path "
                    "to avoid path-resolution ambiguity. Pass an absolute "
                    "path (e.g. /home/ibgateway/warm-state).")
        return
    # Reject absurd paths — / and single-component /root, /etc, /var that
    # would pull in vast amounts of the host filesystem. Legitimate warm
    # state is always under the user's home or a mounted volume.
    suspicious = {"/", "/etc", "/root", "/home", "/var", "/usr", "/tmp"}
    if os.path.realpath(WARM_STATE_DIR) in suspicious:
        log.error(f"GATEWAY_WARM_STATE={WARM_STATE_DIR} resolves to a "
                  f"system directory. Refusing to proceed — this is almost "
                  f"certainly a misconfiguration. Set GATEWAY_WARM_STATE to "
                  f"a dedicated warm-state directory only.")
        return
    # Cap the warm-state size to something reasonable. Gateway's Jts state
    # is typically <10 MB; anything much larger is probably user error or
    # a path pointing at the wrong directory.
    total_bytes = 0
    WARM_STATE_MAX_BYTES = 500 * 1024 * 1024  # 500 MB
    try:
        for root, _, files in os.walk(WARM_STATE_DIR, followlinks=False):
            for f in files:
                try:
                    total_bytes += os.path.getsize(os.path.join(root, f))
                except OSError:
                    pass
                if total_bytes > WARM_STATE_MAX_BYTES:
                    break
            if total_bytes > WARM_STATE_MAX_BYTES:
                break
    except Exception as e:
        log.warning(f"GATEWAY_WARM_STATE: couldn't measure directory size: {e}")
        return
    if total_bytes > WARM_STATE_MAX_BYTES:
        log.error(f"GATEWAY_WARM_STATE={WARM_STATE_DIR} contains "
                  f"more than {WARM_STATE_MAX_BYTES // (1024 * 1024)} MB. "
                  f"Refusing to copy — this is far larger than any real "
                  f"Jts state directory. Check the path.")
        return
    import shutil
    log.info(f"Applying warm state from {WARM_STATE_DIR} → {JTS_CONFIG_DIR} "
             f"({total_bytes // 1024} KB)")
    os.makedirs(JTS_CONFIG_DIR, exist_ok=True)
    for item in os.listdir(WARM_STATE_DIR):
        # Skip log files and Gateway installation directories — those
        # belong to the test container's own version, not the warm state.
        if item.startswith("launcher") and item.endswith(".log"):
            continue
        if item == "ibgateway":
            continue
        src = os.path.join(WARM_STATE_DIR, item)
        dst = os.path.join(JTS_CONFIG_DIR, item)
        try:
            if os.path.isdir(src):
                shutil.copytree(src, dst, dirs_exist_ok=True)
            else:
                shutil.copy2(src, dst)
            log.info(f"  copied {item}")
        except Exception as e:
            log.error(f"  failed to copy {item}: {e}")


def ensure_jts_ini():
    """Ensure a usable jts.ini is in place before Gateway starts.

    Resolution order (highest precedence first):
      1. TWS_SERVER (or TWS_SERVER_PAPER for paper mode) — explicit user
         choice. The controller writes a complete jts.ini with the
         regional server, port routing, AND a SupportsSSL cache entry.
         Overwrites any existing jts.ini (e.g. the empty file that
         run.sh's apply_settings may have left us).
      2. Existing jts.ini (from apply_warm_state(), from run.sh's
         apply_settings rendering a user-provided template, or from a
         mounted volume) — leave alone.
      3. Minimal default — let Gateway figure it out.

    The SupportsSSL cache entry is critical. Without it, Gateway
    re-negotiates SSL with IBKR's misc URLs server on port 4000 on
    every boot, and the negotiation sometimes fails with
    "Remote host terminated the handshake". Pre-populating the cache
    with today's date tells Gateway "SSL is known to work on this
    endpoint, skip negotiation".
    """
    jts_ini = os.path.join(JTS_CONFIG_DIR, "jts.ini")
    os.makedirs(JTS_CONFIG_DIR, exist_ok=True)
    time_zone = os.environ.get("TIME_ZONE", "Etc/UTC")

    if TWS_SERVER:
        import datetime
        cache_date = datetime.datetime.now().strftime("%Y%m%d")
        existed = os.path.exists(jts_ini)
        log.info(f"Writing jts.ini for server {TWS_SERVER} (overwriting={existed})")
        # Do NOT set RemoteHostOrderRouting here. Auth and order routing
        # can be on DIFFERENT IBKR regional servers (e.g. auth on cdc1,
        # orders on ndc1). Gateway discovers the order routing endpoint
        # from the auth server's response and auto-populates
        # RemoteHostOrderRouting in jts.ini after a successful login.
        # Writing it to the same value as Peer (which is what we have)
        # causes a "No Internet connection" retry loop on accounts where
        # the two are different.
        #
        # Bug found via internet research: confirmed by mvberg/ib-gateway-docker
        # and the user's own warm-state jts.ini (Peer=cdc1, RemoteHost=ndc1).
        content = (
            "[IBGateway]\n"
            "WriteDebug=false\n"
            "TrustedIPs=127.0.0.1\n"
            "ApiOnly=true\n"
            "LocalServerPort=4000\n"
            "\n"
            "[Logon]\n"
            f"TimeZone={time_zone}\n"
            "Locale=en\n"
            "displayedproxymsg=1\n"
            "UseSSL=true\n"
            "s3store=true\n"
            "useRemoteSettings=false\n"
            # Pre-populated SSL support cache. Tells Gateway "SSL works
            # on this endpoint, don't re-negotiate" — the missing cache
            # is what causes SSLHandshakeException on cold-start misc
            # URLs requests.
            f"SupportsSSL={TWS_SERVER}:4000,true,{cache_date},false\n"
            "\n"
            "[Communication]\n"
            f"Peer={TWS_SERVER}:4001\n"
            "Region=usr\n"
        )
        with open(jts_ini, "w") as f:
            f.write(content)
        log.info(f"Wrote {jts_ini}")
        return

    if os.path.exists(jts_ini):
        log.info(f"Existing jts.ini at {jts_ini} — leaving in place")
        return

    log.warning("TWS_SERVER not set — writing minimal jts.ini, Gateway will use defaults")
    content = (
        "[IBGateway]\n"
        "WriteDebug=false\n"
        "TrustedIPs=127.0.0.1\n"
        "ApiOnly=true\n"
        "\n"
        "[Logon]\n"
        f"TimeZone={time_zone}\n"
        "Locale=en\n"
        "displayedproxymsg=1\n"
        "UseSSL=true\n"
        "s3store=true\n"
    )
    with open(jts_ini, "w") as f:
        f.write(content)
    log.info(f"Wrote minimal {jts_ini}")


def launch_gateway():
    """Spawn the Gateway JVM via the install4j launcher.

    Sets INSTALL4J_ADD_VM_PARAMS to:
      - inject the java-atk-wrapper jar onto the boot classpath so the
        ATK assistive technology can load
      - override -DjtsConfigDir, because the bundled launcher script has
        a literal unsubstituted `${installer:jtsConfigDir}` placeholder
        that would otherwise route Gateway's writes to /root/Jts (which
        fails for the non-root ibgateway user)
    """
    global _GATEWAY_LAUNCHER_PATH
    launcher = find_gateway_launcher()
    if launcher is None:
        log.error(f"No Gateway launcher found under {TWS_PATH}/ibgateway")
        sys.exit(1)
    _GATEWAY_LAUNCHER_PATH = launcher
    log.info(f"Gateway launcher: {launcher}")
    log.info(f"Gateway config dir (jtsConfigDir): {JTS_CONFIG_DIR}")

    # Apply warm state BEFORE writing the default jts.ini, so a warm
    # jts.ini wins over the default and routes Gateway to the right
    # regional server.
    apply_warm_state()
    ensure_jts_ini()

    env = os.environ.copy()
    # Gateway never reads these; the password reaches it through the agent
    # socket. Keep them out of the JVM's environment.
    for secret in ("TWS_PASSWORD", "TWS_PASSWORD_PAPER", "TWOFACTOR_CODE",
                   "TWOFACTOR_CODE_PAPER", "VNC_SERVER_PASSWORD",
                   "CONTROLLER_COMMAND_SERVER_AUTH_TOKEN"):
        env.pop(secret, None)
    # Java module-access flags required by Gateway's auth and UI code.
    # Without these, reflective access to internal JDK classes fails
    # silently and AuthDispatcher.connect never fires — the auth request
    # is never sent, producing a 20-second silent timeout.
    #
    # These are the same flags IBC's ibcstart.sh passes. The install4j
    # launcher's .vmoptions file does NOT include them; IBC adds them
    # externally. We must do the same via INSTALL4J_ADD_VM_PARAMS.
    #
    # Root cause: Gateway 10.45+ uses reflection into java.desktop and
    # java.base internals for Swing threading, AWT event dispatch, and
    # the auth connection handshake. Java 17's module system blocks
    # this by default. The --add-opens/--add-exports flags grant the
    # access Gateway expects.
    module_access = [
        "--add-opens=java.base/java.util=ALL-UNNAMED",
        "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED",
        "--add-exports=java.base/sun.util=ALL-UNNAMED",
        "--add-exports=java.desktop/com.sun.java.swing.plaf.motif=ALL-UNNAMED",
        "--add-opens=java.desktop/java.awt=ALL-UNNAMED",
        "--add-opens=java.desktop/java.awt.dnd=ALL-UNNAMED",
        "--add-opens=java.desktop/javax.swing=ALL-UNNAMED",
        "--add-opens=java.desktop/javax.swing.event=ALL-UNNAMED",
        "--add-opens=java.desktop/javax.swing.plaf.basic=ALL-UNNAMED",
        "--add-opens=java.desktop/javax.swing.table=ALL-UNNAMED",
        "--add-opens=java.desktop/sun.awt=ALL-UNNAMED",
        "--add-exports=java.desktop/sun.awt.X11=ALL-UNNAMED",
        "--add-exports=java.desktop/sun.swing=ALL-UNNAMED",
        "--add-opens=javafx.graphics/com.sun.javafx.application=ALL-UNNAMED",
        "--add-exports=javafx.media/com.sun.media.jfxmedia=ALL-UNNAMED",
        "--add-exports=javafx.media/com.sun.media.jfxmedia.events=ALL-UNNAMED",
        "--add-exports=javafx.media/com.sun.media.jfxmedia.locator=ALL-UNNAMED",
        "--add-exports=javafx.media/com.sun.media.jfxmediaimpl=ALL-UNNAMED",
        "--add-exports=javafx.web/com.sun.javafx.webkit=ALL-UNNAMED",
        "--add-opens=jdk.management/com.sun.management.internal=ALL-UNNAMED",
    ]
    vm_params = module_access + [
        # v0.5.12: disable the AT-SPI java-atk-wrapper bridge as
        # defense-in-depth. v0.5.13 removed the bridge JAR from the
        # image entirely (so AtkWrapper has no class to load anyway),
        # but we still set the property to an empty value so the JRE's
        # own accessibility-property-file lookup never tries to
        # instantiate it on a base image that ships the JAR pre-installed.
        #
        # Background: the bridge ships an AWT property-change listener
        # (AtkWrapper$5.propertyChange) that fires on every component
        # property update — including JProgressBar.setValue calls from
        # IBKR's "Connecting…" welcome screen during login. The listener's
        # native emitSignal does a JNI re-entry that calls back into
        # AtkObject.hashCode → AtkUtil.invokeInSwing, which posts a
        # FutureTask to the AWT EventQueue and parks waiting on it. The
        # AWT EventQueue itself is meanwhile blocked at
        # AtkWrapper$6.dispatchEvent waiting for monitor entry — and the
        # FutureTask never runs. JTS-Login-14 hangs forever holding the
        # connection-state lock; JTS-CCPListenerS2 can't dispatch the
        # NS_AUTH_START response from IBKR; the controller times out
        # after 20 s and emits a misleading CCP LOCKOUT alert.
        # Verified by SIGQUIT thread dump 2026-04-27.
        #
        # All login-UI interaction (set credentials, click Log In, toggle
        # Live/Paper) goes through the in-JVM gateway-input-agent, which
        # uses pure Swing/AWT (Window.getWindows + SwingUtilities) and
        # does NOT depend on AT-SPI for any of its work.
        "-Djavax.accessibility.assistive_technologies=",
        f"-DjtsConfigDir={JTS_CONFIG_DIR}",
    ]
    if os.path.exists(AGENT_JAR):
        vm_params.append(f"-javaagent:{AGENT_JAR}={AGENT_SOCKET}")
        log.info(f"Loading input agent: {AGENT_JAR} (socket={AGENT_SOCKET})")
        # Clear any stale socket from a previous run
        try:
            os.unlink(AGENT_SOCKET)
        except FileNotFoundError:
            pass
    else:
        log.warning(f"Input agent jar not found at {AGENT_JAR} — text input will fail")
    env["INSTALL4J_ADD_VM_PARAMS"] = " ".join(vm_params)

    # The install4j launcher has a literal unsubstituted placeholder
    # `-DjtsConfigDir=${installer:jtsConfigDir}` that it passes to Java
    # BEFORE our INSTALL4J_ADD_VM_PARAMS. Java uses the FIRST -D for
    # any given property, so our `-DjtsConfigDir=...` override gets
    # ignored and Gateway reads a nonexistent path.
    #
    # Fix: pass `-VjtsConfigDir=<path>` as a command-line argument to
    # the install4j launcher. The `-V` flag sets the installer variable
    # BEFORE the launcher constructs the Java command line, so the
    # `${installer:jtsConfigDir}` placeholder gets correctly substituted
    # to our desired path. Same mechanism for `installerType`.
    launcher_args = [
        launcher,
        f"-VjtsConfigDir={JTS_CONFIG_DIR}",
        "-VinstallerType=standalone",
    ]

    # Diagnostic: capture JVM stdout/stderr to a file so SIGQUIT thread dumps
    # can be read after the fact. Without this, JVM stderr goes to /dev/null
    # and SIGQUIT-triggered thread dumps are lost.
    jvm_console_path = f"/tmp/jvm_console_{TRADING_MODE}.log"
    jvm_console_fd = open(jvm_console_path, "w")
    proc = subprocess.Popen(
        launcher_args,
        env=env,
        stdout=jvm_console_fd,
        stderr=subprocess.STDOUT,
    )
    log.info(f"Gateway PID: {proc.pid} (JVM console -> {jvm_console_path})")
    global _JVM_STARTED_WALL, _logoff_backstop_armed
    global _existing_session_continued_this_jvm, _existing_session_closed_after_continue
    _JVM_STARTED_WALL = datetime.now()
    _logoff_backstop_armed = False
    _existing_session_continued_this_jvm = False
    _existing_session_closed_after_continue = False
    return proc


# ── Dialog handlers ─────────────────────────────────────────────────────

def handle_login(app):
    """Drive the login dialog: select trading mode, type credentials, click Log In.

    v0.5.12: removed the pyatspi tree-walking path. The AT-SPI bridge is
    disabled in the JVM (see ``launch_gateway`` — the
    ``-Djavax.accessibility.assistive_technologies=`` flag) to prevent the
    AtkWrapper$5.propertyChange + AtkUtil.invokeInSwing deadlock that hung
    JTS-Login-14 during the welcome-screen JProgressBar update. v0.5.14
    physically removed the dead helpers (find_descendant / wait_for /
    get_states / click(node) / _read_text / set_text(node, ...) /
    _dump_tree) and the gi.repository.Atspi import. All login-UI
    interaction goes through the in-JVM ``gateway-input-agent`` socket —
    pure Swing/AWT, no AT-SPI callbacks, no JNI re-entrancy, no deadlock.

    The ``app`` argument is now a placeholder kept only for caller
    compatibility (attempt_inplace_relogin, do_restart_in_place,
    attempt_reauth all still pass an `app` reference through). The agent
    discovers the login frame internally via ``findLoginFrame`` (any
    showing Window containing a JPasswordField).
    """
    log.info("Waiting for login dialog (JPasswordField in showing Window)")
    if not agent_wait_login_frame(timeout_ms=120_000):
        log.error("Login dialog never appeared (no JPasswordField in any showing Window)")
        return False
    log.info("Login dialog detected")

    # Trading mode — must select BEFORE typing credentials, because Gateway
    # may reset the form when toggling. The login frame's title is
    # "IBKR Gateway" (matched via title-substring); within that window the
    # agent's JCHECK looks for a JToggleButton named "Live Trading" or
    # "Paper Trading" and clicks it iff the state needs to change. If the
    # window match fails (different Gateway version / locale) we log a
    # warning and continue — the warm-state jts.ini already records the
    # last-used mode so a missed toggle usually leaves Gateway with the
    # right selection anyway. If it doesn't, IBKR rejects the credentials
    # and we recover via the credential-rejection path.
    target = "Paper Trading" if TRADING_MODE == "paper" else "Live Trading"
    if agent_jcheck("IBKR Gateway", target, True):
        log.info(f"Trading mode set to {target} via agent")
    else:
        log.warning(f"agent_jcheck for {target!r} did not succeed — relying "
                    "on warm-state jts.ini for trading-mode selection")

    # Username / password via the in-JVM role-based lookup. SETTEXT_LOGIN_*
    # commands match on Swing type (first editable non-password / first
    # JPasswordField on the login frame), so they survive accessible-name
    # drift across login attempts (JComboBox autocomplete editor with null
    # AccessibleName, etc).
    user_ok = agent_settext_login_user(USERNAME)
    if not user_ok and not TEST_MODE:
        return False

    pw_ok = agent_settext_login_password(PASSWORD)
    if not pw_ok and not TEST_MODE:
        return False

    if TEST_MODE and (not user_ok or not pw_ok):
        log.warning("set_text reported failure but TEST_MODE — proceeding to click Log In so we can observe Gateway response")

    # Log In button. Gateway renames the button based on trading mode:
    #   Live Trading  → "Log In"
    #   Paper Trading → "Paper Log In"
    # Try the label this mode is expected to use first, and probe quietly:
    # before v0.9.0 every paper login emitted a spurious
    # `agent CLICK 'Log In': ERR not_found` at ERROR level immediately
    # before succeeding on the second label, which false-positived any
    # "ERROR means page someone" rule.
    expected, fallback = _login_button_labels()
    if not (agent_click(expected, quiet=True) or agent_click(fallback)):
        log.error("Log In / Paper Log In button click failed via agent")
        # Diagnostic: dump the login window's component tree via the agent.
        # This is an ERROR-level (always-emitted) dump of the login frame,
        # which holds the just-typed password. v0.6.3: the agent masks
        # JPasswordField contents at the source, so the password no longer
        # reaches this dump. We additionally run every line through
        # _redact_logs here as defense-in-depth (it strips IBKR DU/U
        # account numbers that can appear in window titles) and because
        # this is exactly the dump a user copies into a bug report.
        try:
            log.error("=== agent_window('IBKR Gateway') dump ===")
            for line in agent_window("IBKR Gateway").split("\n"):
                if line and line not in ("OK", "END"):
                    log.error(f"  {_redact_logs(line)}")
            log.error("=== end window dump ===")
        except Exception as e:
            log.error(f"  agent_window dump failed: {type(e).__name__}: {e}")
        return False

    log.info("Log In clicked successfully")
    return True


# Password-expiry dialog detection. Gateway/TWS surfaces these after a
# successful login when the account's password is within IBKR's
# rotation window. Two known variants:
#   "Your password will expire in N days. Please change it ..."
#   "Your password has expired. You must change it ..."
# The first is informational (login still proceeded); the second is
# blocking (login can't complete until rotation, which has to happen
# in IBKR's web portal — we can't automate that side). Either way the
# right thing is to emit a stable grep-contract alert that distinguishes
# the two so external monitoring can notify the operator *before* the
# account locks out, and escalate differently once it has.
# The outer caller (handle_post_login_dialogs) gates on the dump already
# containing "password" plus "expire"/"expired"; these regexes only have
# to classify the wording.
_PASSWORD_EXPIRED_MATCH = re.compile(r"has\s+expired", re.IGNORECASE)
_PASSWORD_WARNING_MATCH = re.compile(
    r"(?:will\s+expire|expires\s+in\s+\d+\s+day)",
    re.IGNORECASE,
)
_PASSWORD_EXPIRY_DAYS = re.compile(
    r"expire(?:s)?\s+in\s+(\d+)\s+day",
    re.IGNORECASE,
)


def _detect_password_expiry(dump):
    """Parse a window dump for password-expiry wording.

    Returns ``(matched, status, days_remaining)``:
      - ``matched`` is True iff the dump contains expiry wording.
      - ``status`` is ``"expired"`` (already-rotated-past, login blocked)
        or ``"warning"`` (advance notice, login still proceeds). ``None``
        when ``matched`` is False.
      - ``days_remaining`` is the integer extracted from "expire(s) in N
        day(s)" if present, else ``None``. Always ``None`` for
        ``"expired"`` status — the "has expired" variant doesn't carry a
        day count.
    """
    if not dump:
        return False, None, None
    if _PASSWORD_EXPIRED_MATCH.search(dump):
        return True, "expired", None
    if _PASSWORD_WARNING_MATCH.search(dump):
        m = _PASSWORD_EXPIRY_DAYS.search(dump)
        return True, "warning", int(m.group(1)) if m else None
    return False, None, None


# Credential-rejection modal wording. Gateway surfaces a modal such as
# "Connection to server failed: Invalid username or password. Please check
# the Caps Lock key; passwords are case sensitive." when IBKR rejects the
# supplied login. This is distinct from the launcher.log handshake-timeout
# diagnosis in _diagnose_login_failure and from the in-JVM relogin marker
# strings; recognizing the modal lets us emit the ALERT_LOGIN_FAILED
# grep-contract token at the initial-login stage instead of letting the
# dialog fall through unrecognized. The alternation also covers the
# two-word "user name" spelling some IBKR builds use.
_BAD_CREDENTIALS_MATCH = re.compile(
    r"(?:"
    r"(?:invalid|incorrect)\s+user\s?name\s+or\s+password"
    r"|user\s?name\s+or\s+password\s+(?:is\s+|was\s+)?(?:invalid|incorrect)"
    r"|credentials?\s+(?:were\s+|was\s+|are\s+|is\s+)?(?:rejected|invalid|incorrect)"
    r")",
    re.IGNORECASE,
)


def _detect_bad_credentials(dump):
    """Return True when a dialog reports the login credentials were
    rejected (Gateway's "Invalid username or password" modal and close
    variants).

    The caller emits ALERT_LOGIN_FAILED, dismisses the dialog and halts
    (``_halt_credentials_rejected``): this is IBKR saying the password is
    wrong, and every retry counts towards locking the account.
    """
    if not dump:
        return False
    return bool(_BAD_CREDENTIALS_MATCH.search(dump))


def _halt_credentials_rejected(evidence="Gateway reported the username or "
                                        "password as invalid",
                                if_right=""):
    """Stop logging in once IBKR has rejected the credentials.

    Until v0.14.0 a rejection was only an alert and the login was retried
    (a v0.5.0 non-goal, for fear of false positives). Since v0.12.0 the
    controller relaunches a failed login three times and then restarts
    the container, so a wrong password was retried indefinitely, and IBKR
    locks an account after repeated failures. Halting keeps the JVM and
    VNC up, like the 2FA halt; Gateway does not retry a rejected login by
    itself.
    """
    _halt_for_operator(
        f"{evidence}; see ALERT_LOGIN_FAILED above. Fix TWS_USERID / "
        "TWS_PASSWORD (or the _PAPER variants), then restart the container"
        f"{if_right}")


def _latest_login_rejected_credentials():
    """True when the most recent auth attempt in launcher.log carries the
    bad-credentials fingerprint: IBKR answered the handshake
    (NS_AUTH_START) and the CCP auth timer then expired.

    Scoped to the text after the last "Authenticating" line, so lines from
    an earlier attempt cannot combine with a later one's —
    ``_diagnose_login_failure`` reads the whole file and can. A weaker
    signal than Gateway's own rejection dialog, so recovery halts only
    when two consecutive attempts show it, and ignores it inside IBKR's
    daily maintenance window, when a server-side stall is the likelier
    explanation (field report 2026-10-02).
    """
    path = os.path.join(JTS_CONFIG_DIR, "launcher.log")
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            f.seek(0, 2)
            f.seek(max(0, f.tell() - 65536))
            tail = f.read()
    except OSError:
        return False
    start = tail.rfind("Authenticating")
    if start < 0:
        return False
    latest = tail[start:]
    return ("NS_AUTH_START" in latest
            and "AuthTimeoutMonitor-CCP: Timeout!" in latest)


def handle_post_login_dialogs(app):
    """Inspect any modal dialog that appears right after the Log In click,
    handle the ones we recognize, and leave the rest alone.

    The 'Gateway' modal that we previously observed is a
    "Connecting to server..." progress dialog — we should NOT click OK on
    it because that cancels the login. Other modals (existing-session-
    detected, EULA acknowledgement) DO need clicking.

    Strategy:
      1. Wait briefly for any dialog to render
      2. For each modal, dump its full content via the agent's WINDOW
         command and log it for diagnostics
      3. Recognize known dialogs by body text and click the right button
      4. Leave unknown dialogs alone — the downstream waits will catch
         dialog dismissal naturally
    """
    log.info("Inspecting post-login dialogs")
    # Poll for dialog appearance. The 'Existing session detected' dialog
    # can take 3–5 seconds to show up after Log In is clicked (Gateway
    # has to do a network round-trip first). Waiting 2s was enough for
    # the "no dialog" case but missed late-arriving modals.
    modal_dialogs = []
    deadline = time.monotonic() + 6
    while time.monotonic() < deadline:
        time.sleep(0.5)
        windows = agent_windows()
        modal_dialogs = [(t, title) for t, title, modal in windows if modal]
        if modal_dialogs:
            break
    if not modal_dialogs:
        log.info("No modal dialogs after login")
        return True

    log.info(f"Modal dialog(s) detected: {[(t, _redact_logs(title)) for t, title in modal_dialogs]}")
    for type_, title in modal_dialogs:
        dump = agent_window(title)
        # Window dumps are big — emit at debug level so they only
        # appear when the user explicitly asks for them with
        # CONTROLLER_DEBUG=1. Redact account numbers defensively;
        # even at debug level, users share logs when asking for help.
        log.debug(f"=== Window dump: {_redact_logs(repr(title))} ===")
        for line in dump.split("\n"):
            if line and line not in ("OK", "END"):
                log.debug(f"  {_redact_logs(line)}")
        log.debug("=== End window dump ===")

        # Recognize known dialogs by body text content
        body_lower = dump.lower()
        if "existing session" in body_lower or "another session" in body_lower:
            log.info("Recognized: existing-session-detected dialog")
            if not handle_existing_session_dialog():
                log.error("Existing-session dialog handling failed")
                return False
        elif "password" in body_lower and (
                "expire" in body_lower or "expired" in body_lower):
            matched, status, days = _detect_password_expiry(dump)
            if matched:
                if status == "expired":
                    log.error(
                        f"ALERT_PASSWORD_EXPIRED status=expired "
                        f"mode={TRADING_MODE} "
                        f"suggested_action=\"password has expired; "
                        f"rotate in IBKR Account Settings before login "
                        f"will succeed again, then update TWS_PASSWORD\"")
                elif days is not None:
                    log.error(
                        f"ALERT_PASSWORD_EXPIRED status=warning "
                        f"mode={TRADING_MODE} "
                        f"days_remaining={days} "
                        f"suggested_action=\"rotate IBKR password in "
                        f"Account Settings within {days} days to avoid "
                        f"lockout; update TWS_PASSWORD after rotation\"")
                else:
                    log.error(
                        f"ALERT_PASSWORD_EXPIRED status=warning "
                        f"mode={TRADING_MODE} "
                        f"suggested_action=\"rotate IBKR password soon; "
                        f"dialog didn't report remaining days — check "
                        f"IBKR Account Settings for the exact date, then "
                        f"update TWS_PASSWORD after rotation\"")
                # IBC-compat: dismiss so the warning variant doesn't
                # block the rest of the post-login flow. The blocking
                # "already expired" variant will re-appear or the login
                # will fail downstream; either way we've emitted the
                # alert.
                dismissed = False
                for btn in ("OK", "Continue", "Acknowledge", "Close"):
                    if btn in dump and agent_click_in_window(title, btn):
                        log.info(f"Dismissed password-expiry dialog via '{btn}'")
                        dismissed = True
                        break
                if not dismissed:
                    log.warning(
                        "Password-expiry dialog detected but no known "
                        "dismiss button present; leaving dialog in place")
            else:
                log.info("Unrecognized 'password' dialog — leaving in place")
        elif _detect_bad_credentials(dump):
            log.info("Recognized: credential-rejection dialog")
            log.error(
                f"ALERT_LOGIN_FAILED mode={TRADING_MODE} "
                f"reason=\"bad-credentials\" "
                f"suggested_action=\"Gateway surfaced a credential-"
                f"rejection modal; verify TWS_USERID / TWS_PASSWORD "
                f"(or _PAPER variants) and update env if password "
                f"was rotated in IBKR Account Settings\"")
            dismissed = False
            for btn in ("OK", "Close", "Continue"):
                if btn in dump and agent_click_in_window(title, btn):
                    log.info(f"Dismissed credential-rejection dialog via '{btn}'")
                    dismissed = True
                    break
            if not dismissed:
                log.warning(
                    "Credential-rejection dialog detected but no known "
                    "dismiss button present; leaving dialog in place")
            # Retrying a password IBKR just rejected only counts towards
            # locking the account.
            _halt_credentials_rejected()
        else:
            log.info(f"Unrecognized modal — leaving in place to let Gateway flow proceed")

    return True


# Ping-pong mitigation: track recent 'Continue Login' clicks so we can
# detect two containers fighting for the same IBKR account and back off
# instead of flapping forever. Module-level ring buffer of monotonic
# click timestamps, bounded in size.
_existing_session_click_times = []
_EXISTING_SESSION_BACKOFF_WINDOW = 300.0  # 5-minute window
_EXISTING_SESSION_BACKOFF_THRESHOLD = 5   # clicks in the window that trigger backoff
_EXISTING_SESSION_BACKOFF_SLEEP = 60.0    # seconds to wait once tripped
# Takeovers ("Continue Login") this process has made, kept across
# relaunches. Two deployments on one username with TOTP and the primary
# action would otherwise take the session from each other every few
# minutes with nobody involved, feeding IBKR's CCP limiter (security
# review 2026-10-12): the backoff above counts clicks in 5 minutes, and a
# relogin cycle takes longer than that.
_existing_session_takeovers = []
_EXISTING_SESSION_TAKEOVER_LIMIT = 3
_EXISTING_SESSION_TAKEOVER_WINDOW = 7200.0
# IBC's rule, which IBKR's prompt is built around: IBKR asks the new login
# whether to continue, which logs the other session out. The new login
# continues once; if IBKR asks again while the same Gateway is running,
# the other session logged back in, so someone is using it: give way
# rather than take it again. Reset whenever a new Gateway JVM starts, as
# IBC's flag reset with each TWS/Gateway start. The takeover limit above
# is the backstop for takeovers spread across separate launches.
_existing_session_continued_this_jvm = False
# Set once the prompt we continued past has been seen to close. CLICK_IN_WIN
# returns before Swing processes the click, so a poll right after it can
# still list the same dialog; only a prompt that comes back after closing
# means the other session logged in again.
_existing_session_closed_after_continue = False


def _existing_session_shown():
    """True / False, or None when the agent didn't answer: an error must
    never read as "closed", or a re-poll of the open dialog would look like
    IBKR asking again."""
    try:
        raw = _agent_multiline("WINDOWS")
    except Exception:
        return None
    if not raw.startswith("OK"):
        return None
    return any(modal and ("existing session" in t.lower() or "another session" in t.lower())
               for _w, t, modal in _parse_windows(raw))


def _wait_existing_session_closed(timeout=10.0):
    """True once the existing-session dialog is seen not showing."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if _existing_session_shown() is False:
            return True
        time.sleep(0.5)
    return False


def _note_existing_session_absent():
    """Called by the polling loops when their window list has no
    existing-session dialog: the prompt we continued past has closed, so a
    later one is IBKR asking again."""
    global _existing_session_closed_after_continue
    if _existing_session_continued_this_jvm:
        _existing_session_closed_after_continue = True


def _log_existing_session_prompt():
    """Log what IBKR's dialog says (which session, connected when), so an
    operator sees exactly what was asked. Public addresses are masked."""
    try:
        lines = [text for title, text in agent_labels()
                 if "existing session" in title.lower()]
    except Exception:
        return
    for text in lines[:6]:
        log.info(_redact_logs(f"  IBKR: {text}"))


def handle_existing_session_dialog():
    """Click the appropriate button on Gateway's 'Existing session detected'
    dialog based on the EXISTING_SESSION_DETECTED_ACTION env var.

    IBC supports four actions for this dialog (from the IBC documentation):
      primary          — accept this session as the primary, kicking out
                         the existing one (the most common automation
                         setting; what most production users want)
      primaryoverride  — same as primary but always overrides
      secondary        — connect as secondary, leaving the existing
                         session intact
      manual           — leave the dialog up; the user must click

    The modern Gateway dialog (verified on 10.45.1c) has buttons
    "Continue Login" (= primary / accept and kick the other session)
    and "Cancel" (abort this login, leave the other session alone).
    There is no separate "connect as secondary" button — the workflow
    is "use the already-running session, don't connect here" which maps
    onto Cancel.

    For headless operation, 'primary' is the typical default.

    Ping-pong backoff: if this handler fires more than
    _EXISTING_SESSION_BACKOFF_THRESHOLD times in
    _EXISTING_SESSION_BACKOFF_WINDOW seconds, we're probably in a fight
    with another container that keeps reconnecting as the same account.
    We sleep for _EXISTING_SESSION_BACKOFF_SLEEP seconds before
    clicking again to give the user a chance to intervene without
    flapping forever.
    """
    action = os.environ.get("EXISTING_SESSION_DETECTED_ACTION", "primary").lower()
    takes_over = action in ("primary", "primaryoverride")
    global _existing_session_continued_this_jvm, _existing_session_closed_after_continue
    if (takes_over and _existing_session_continued_this_jvm
            and not _existing_session_closed_after_continue):
        # Still the prompt we already answered (CLICK_IN_WIN is async): wait
        # for it to close. Clicks nothing, so it doesn't count below.
        _existing_session_closed_after_continue = _wait_existing_session_closed()
        return True
    log.info(f"existing-session dialog: action={action}")

    # Ping-pong detection
    now = time.monotonic()
    _existing_session_click_times[:] = [
        t for t in _existing_session_click_times
        if now - t < _EXISTING_SESSION_BACKOFF_WINDOW
    ]
    _existing_session_click_times.append(now)
    if len(_existing_session_click_times) >= _EXISTING_SESSION_BACKOFF_THRESHOLD:
        log.warning(f"existing-session: handled this dialog "
                    f"{len(_existing_session_click_times)} times in "
                    f"{int(_EXISTING_SESSION_BACKOFF_WINDOW)}s — another "
                    f"container or app is probably reconnecting as the same "
                    f"account. Backing off for {int(_EXISTING_SESSION_BACKOFF_SLEEP)}s.")
        time.sleep(_EXISTING_SESSION_BACKOFF_SLEEP)
        _existing_session_click_times.clear()

    # Ordered candidates: try the most specific/modern button text first,
    # then fall back to older labels for compatibility with older Gateway
    # versions that may have had different button naming.
    button_candidates_by_action = {
        "primary":         ["Continue Login", "Primary", "Yes", "OK", "Continue"],
        "primaryoverride": ["Continue Login", "Primary", "Yes", "OK", "Continue"],
        "secondary":       ["Cancel", "Secondary", "No"],
        "manual":          [],  # do nothing
    }
    candidates = button_candidates_by_action.get(action, ["OK"])
    _log_existing_session_prompt()
    if takes_over and _existing_session_continued_this_jvm:
        log.warning("existing-session: IBKR asked again after this Gateway "
                    "already continued once, so the other session logged "
                    "back in and is in use; giving way")
        for cand in ("Cancel", "Exit Application"):
            if agent_click_in_window("Existing session", cand):
                break
        log.error(
            f"ALERT_EXISTING_SESSION_CONFLICT mode={TRADING_MODE} "
            f"reason=\"the other session logged back in after this one took "
            f"over; gave way\" "
            f"remediation=\"stop the other session (another container, TWS, "
            f"Client Portal, IBKR Mobile), then restart this container\"")
        _halt_for_operator(
            "another session of this username is in use; see "
            "ALERT_EXISTING_SESSION_CONFLICT above")
        return False
    if takes_over:
        now = time.monotonic()
        _existing_session_takeovers[:] = [
            t for t in _existing_session_takeovers
            if now - t < _EXISTING_SESSION_TAKEOVER_WINDOW]
        if len(_existing_session_takeovers) >= _EXISTING_SESSION_TAKEOVER_LIMIT:
            log.error(
                f"ALERT_EXISTING_SESSION_CONFLICT mode={TRADING_MODE} "
                f"takeovers={len(_existing_session_takeovers)} "
                f"window_seconds={int(_EXISTING_SESSION_TAKEOVER_WINDOW)} "
                f"reason=\"another session keeps logging in as this username\" "
                f"remediation=\"stop the other session (another container, "
                f"TWS, Client Portal, IBKR Mobile), then restart this "
                f"container\"")
            _halt_for_operator(
                "another session keeps logging in as this username; see "
                "ALERT_EXISTING_SESSION_CONFLICT above")
            return False
    if not candidates:
        log.info("action=manual — leaving dialog for user")
        return True

    # Scope the click to the existing-session dialog window specifically,
    # via CLICK_IN_WIN. Using the global agent_list + agent_click would
    # risk hitting a button with the same label in the main window
    # (especially in dual-mode containers).
    title_substr = "Existing session"
    for cand in candidates:
        log.info(f"Clicking existing-session button {cand!r} in {title_substr!r} window")
        if agent_click_in_window(title_substr, cand):
            if takes_over:
                _existing_session_takeovers.append(time.monotonic())
                _existing_session_continued_this_jvm = True
                _existing_session_closed_after_continue = _wait_existing_session_closed()
            return True
    log.error(f"existing-session: none of {candidates!r} worked as a "
              f"button label in the dialog")
    return False


def _resolve_twofa_device(explicit, has_totp):
    """Resolve which second-factor method we expect Gateway's 2FA dialog
    to be requesting. Explicit ``TWOFA_DEVICE`` wins; otherwise default to
    the Mobile Authenticator (TOTP) method when a TOTP secret is
    configured, else IB Key. IBC-compatible (``SecondFactorDevice``).
    """
    explicit = (explicit or "").strip()
    if explicit:
        return explicit
    return "Mobile Authenticator app" if has_totp else "IB Key"


def _twofa_selector_present(window_dump):
    """True if a ``WINDOW`` component-tree dump of the Second Factor
    window shows the multi-method *device-selector* variant (issue #20):
    a "Select second factor device" heading over a JList of methods.

    Gateway 10.45.x has two account-dependent shapes for this dialog —
    the selector, and a code dialog pre-defaulted to one method
    (issue #7 spike). The selector's heading is a JTextArea, which
    ``agent_labels()`` (JLabel-only) cannot see, so detection needs the
    full ``agent_window()`` dump. Tolerates ``None``/"" (agent error
    paths return empty strings).
    """
    return "Select second factor device" in (window_dump or "")


# Window-title fragments that positively identify IBKR's passkey /
# WebAuthn login flow. On a passkey account Gateway launches an in-app
# browser (jxbrowser) that runs the WebAuthn ceremony and expects a
# hardware security key, instead of showing the Second Factor dialog we
# drive. IBKR forced this on Hong Kong and Japan accounts (2026-08) and
# may widen it. These fragments are deliberately high-specificity —
# none appears in a normal Gateway login window set — so a match means
# we are NOT in a TOTP flow. PROVISIONAL: assembled without a live
# passkey account (issue #22); a real WINDOW dump may refine them. A
# non-match is not proof it isn't passkey (the jxbrowser window title
# may be generic) — the timeout path carries a fallback diagnostic.
_PASSKEY_FLOW_TITLE_FRAGMENTS = (
    "passkey",
    "webauthn",
    "security key",
    "jxbrowser",
)


def _detect_passkey_flow(windows):
    """Return the passkey/WebAuthn window-title fragment that matched, or
    ``None``. ``windows`` is ``agent_windows()`` output — a list of
    ``(type, title, modal)`` tuples. Lets ``handle_2fa`` fail loudly when
    IBKR routed the login through a passkey ceremony a headless container
    can't satisfy (issue #22), rather than waiting out the 2FA timeout
    for a dialog that will never appear."""
    for _type, title, _modal in windows:
        tl = (title or "").lower()
        for frag in _PASSKEY_FLOW_TITLE_FRAGMENTS:
            if frag in tl:
                return frag
    return None


def _passkey_prompt_present(window_dump):
    """Gateway can embed passkey 2FA in a generically titled dialog."""
    return "use your passkey device" in " ".join((window_dump or "").casefold().split())


def _handle_passkey_prompt(title):
    """Return None if absent, or whether an Authenticate click was accepted.

    Gateway 10.50's J is a JButton subclass. Use the original agent's
    CLICK_IN_WIN/doClick path, never guessed coordinates. LIST exposes raw
    accessible names, whereas WINDOW strips HTML and collapses whitespace.
    A successful request only starts the ceremony; it is not login success.
    """
    dump = agent_window(title)
    if not _passkey_prompt_present(dump):
        return None
    log.info("Passkey prompt detected in %r: Use your Passkey device", title)
    if not _PASSKEY_AUTHENTICATE:
        # Same reason string v0.8.1 emits for the browser-window variant,
        # so monitors keyed on it keep matching. Without this gate the
        # dialog would fall through to TOTP handling and the code would
        # be typed into a field that isn't there.
        log.error(
            f"ALERT_2FA_FAILED mode={TRADING_MODE} "
            f"reason=\"passkey/WebAuthn 2FA flow - unattended login not supported\" "
            f"remediation=\"set PASSKEY_AUTHENTICATE=yes to have the "
            f"controller press Authenticate, and run an authenticator "
            f"alongside the container to complete the ceremony (README, "
            f"Passkey section)\"")
        return _fail_2fa_needs_operator()
    # WINDOW can dump multiple matches, and CLICK_IN_WIN would click the
    # first matching button it finds (modal dialogs first).
    if sum(line.startswith("=== window=") for line in dump.splitlines()) > 1:
        log.error("Multiple windows match %r; refusing an ambiguous passkey click", title)
        return False

    labels = ("Authenticate >", "Authenticate", "Authenticate ›")
    attempted = set()

    def click(label):
        attempted.add(label)
        log.info("Passkey Authenticate candidate (raw): %r", label)
        if not agent_click_in_window(title, label):
            return False
        # Returning True from here means handle_2fa reports success as
        # soon as the click is accepted, before the ceremony completes.
        # If the external authenticator never answers, the failure
        # surfaces downstream as the API-port wait timing out rather than
        # as a 2FA failure. That hand-off is deliberate: the controller
        # has done the only part it can do.
        log.info(
            "Passkey Authenticate click accepted by agent; no further clicks "
            "in this attempt. WebAuthn completion is outside this handler.")
        # The dialog can remain open during the ceremony. Its status/button
        # state is more useful than assuming the whole window must disappear.
        time.sleep(0.5)
        after = agent_window(title)
        if after and after.startswith("OK"):
            log.info("Passkey post-click dialog state:\n%s", after)
        else:
            log.warning("Passkey click accepted, but UI verification unavailable")
        return True

    for label in labels:
        if click(label):
            return True

    # These names are global to this JVM, but activation stays scoped to the
    # detected dialog. Preserve the raw spelling for exact agent matching.
    _, names = agent_list("Authenticate")
    candidates = []
    for name in sorted(names):
        if any(c in name for c in "\r\n"):
            continue  # The agent protocol is line-oriented.
        normalised = " ".join(re.sub(r"<[^>]+>", " ", name).split())
        if normalised in labels:
            log.info("Passkey LIST button name (raw): %r", name)
            if name not in attempted:
                candidates.append(name)
    if candidates:
        # Revalidate the body before using identifiers from the global LIST.
        if not _passkey_prompt_present(agent_window(title)):
            log.warning("Passkey prompt changed before raw-name retry; not clicking")
            return False
        for name in candidates:
            if click(name):
                return True

    log.error(
        'ALERT_2FA_FAILED mode=%s reason="passkey Authenticate lookup failed"',
        TRADING_MODE)
    log.error("Passkey matching accessible names (escaped): %r", sorted(names))
    log.error("Passkey dialog state after failed lookup:\n%s",
              _redact_logs(agent_window(title)))
    return False


def _twofa_requested_method(labels, window_substr=None):
    """From ``agent_labels()`` output (a list of ``(window_title, text)``
    tuples) return Gateway's 2FA method-prompt label — IBKR phrases it as
    "Enter <method> code" — or ``None`` if no such label is present.

    On a multi-method account Gateway pre-defaults the Second Factor
    dialog to ONE method and shows this prompt; reading it tells us which
    method the dialog is actually expecting.

    ``window_substr`` scopes the search to labels whose window title
    contains it. Pass it the 2FA window substring. NOTE: this is the
    window-title scope — do NOT rely on ``agent_labels(<substr>)`` for
    it, because that filters by label TEXT, not window title (which
    silently dropped the prompt in v0.7.0-rc1; the prompt text doesn't
    contain "Second Factor"). Always call ``agent_labels()`` with no
    argument and scope here.
    """
    for w, text in labels:
        if window_substr and window_substr not in (w or ""):
            continue
        t = (text or "").strip()
        tl = t.lower()
        if tl.startswith("enter ") and tl.endswith(" code"):
            return t
    return None


def _twofa_method_mismatch(prompt, desired_device):
    """Return True only if ``prompt`` ("Enter <method> code") positively
    names a method OTHER than ``desired_device``.

    Deliberately lenient — returns False whenever we can't be sure (no
    recognizable prompt, empty desired, or a match), so single-method
    accounts and any dialog wording we don't recognize keep the existing
    "just type the code" behavior. We only block when we POSITIVELY
    detect the dialog is asking for a different method than the one we
    can satisfy, to avoid typing a TOTP into the wrong method (issue #7).
    """
    if not prompt:
        return False
    d = (desired_device or "").strip().lower()
    if not d:
        return False
    p = prompt.lower()
    if d in p:
        return False
    # Fall back to the distinctive head token (e.g. "mobile authenticator"),
    # so minor wording drift in the env value still matches the prompt.
    head = " ".join(d.split()[:2])
    if head and head in p:
        return False
    return True


# Set when a 2FA failure needs a human to change something — the account's
# enabled methods, TWOFA_DEVICE, PASSKEY_AUTHENTICATE — rather than another
# login attempt. main() halts on these instead of exiting: exiting hands the
# container back to Docker's restart policy, which re-runs the identical
# login, and a configuration failure fails identically every time. That turns
# one misconfigured account into a login generator pointed at IBKR's rate
# limiter (issues #20, #37).
_twofa_needs_operator = False


def _fail_2fa_needs_operator():
    """Mark this 2FA failure as one only a human can clear.

    Returns False so callers can ``return _fail_2fa_needs_operator()`` in
    place of ``return False``.
    """
    global _twofa_needs_operator
    _twofa_needs_operator = True
    return False


def handle_2fa(app):
    """Handle Gateway's Second Factor Authentication dialog.

    Passkey prompts are detected by their body text and Authenticate is
    activated once via the agent; completing WebAuthn remains out of scope.
    Otherwise supports these modes:

    1. **TOTP mode** (TWOFACTOR_CODE set): the controller generates a
       TOTP code and types it into the dialog's text field, then clicks
       OK. Fully automated, no human interaction.

    2. **IB Key push mode** (TWOFACTOR_CODE NOT set, but the 2FA dialog
       appears anyway): IBKR sends a push notification to the user's
       phone via the IB Key mobile app. The controller detects the
       dialog, logs "Waiting for IB Key mobile approval — approve on
       your phone", and polls for the dialog to disappear (which means
       the user approved) or the API port to open. No text entry, no
       click — just waiting for the human to approve.

       This is how ibctl handles push 2FA too: the Swing tree shows
       the dialog, the user acts on their phone, the dialog closes
       itself, and the automation proceeds.

    3. **No 2FA at all** (neither TWOFACTOR_CODE set nor dialog appears):
       common for paper accounts and autorestart-token sessions. The
       handler polls briefly for the dialog; if the API port opens
       first, it skips out.

    Selectors captured from real-credential live-mode testing:
      - Window title: "Second Factor Authentication" (JDialog, modal)
      - TOTP body label: "Enter Mobile Authenticator app code"
      - IB Key body label: typically mentions "IB Key" or "approval"
      - Input field (TOTP): first JTextComponent inside the window
      - Submit: button with text="OK"
      - Cancel: button with text="Cancel"

    Early exit: if the API port opens before the 2FA dialog appears,
    Gateway authenticated without 2FA — skip the wait entirely.
    """
    # Even without TOTP_SECRET, we still enter the polling loop so we
    # can detect and wait for IB Key push approval. The only difference
    # is: with TOTP_SECRET we type the code; without it we just wait
    # for the dialog to disappear.
    global _twofa_needs_operator
    _twofa_needs_operator = False

    ib_key_mode = not TOTP_SECRET
    if ib_key_mode:
        log.info("No TWOFACTOR_CODE set — will watch for IB Key push dialog if it appears")

    # v0.7.0 (issue #7): the method we can satisfy. On a multi-method
    # account Gateway pre-defaults the 2FA dialog to one method; we read
    # the dialog's "Enter <method> code" prompt and only type the TOTP if
    # it's asking for this method, rather than mis-typing into the wrong
    # one. Defaults to Mobile Authenticator in TOTP mode (IBC-compatible
    # TWOFA_DEVICE override).
    twofa_device = _resolve_twofa_device(os.environ.get("TWOFA_DEVICE"),
                                         bool(TOTP_SECRET))

    # IBC-compat 2FA timeout configuration:
    #   TWOFA_EXIT_INTERVAL   — seconds to wait for the dialog (default 120)
    #   TWOFA_TIMEOUT_ACTION  — what to do on timeout: 'exit' (default,
    #                           controller exits non-zero), 'restart'
    #                           (call do_restart_in_place), or 'none'
    #                           (fall through to wait_for_api_port as we
    #                           did previously)
    #   RELOGIN_AFTER_TWOFA_TIMEOUT — yes/no; if yes, after a TWOFA
    #                           timeout drive handle_login again before
    #                           giving up
    try:
        wait_seconds = int(os.environ.get("TWOFA_EXIT_INTERVAL", "120"))
    except ValueError:
        log.warning("TWOFA_EXIT_INTERVAL not an integer; using 120")
        wait_seconds = 120
    timeout_action = os.environ.get("TWOFA_TIMEOUT_ACTION", "none").strip().lower()
    relogin_after_timeout = _coerce_yes_no(
        os.environ.get("RELOGIN_AFTER_TWOFA_TIMEOUT", ""))

    TWOFA_WINDOW_SUBSTR = "Second Factor"
    api_port = api_port_for_mode()
    start = time.monotonic()
    deadline = start + wait_seconds
    last_log = 0.0
    last_windows = None

    log.info(f"Waiting for 2FA dialog (window title contains {TWOFA_WINDOW_SUBSTR!r}) "
             f"for up to {wait_seconds}s")
    while time.monotonic() < deadline:
        # Early exit: API port already open → login complete, no 2FA needed
        if is_api_port_open(api_port):
            log.info(f"API port {api_port} already open — no 2FA dialog needed")
            return True

        windows = agent_windows()

        # Log window changes for diagnostics
        if windows != last_windows:
            log.info(_redact_logs(f"2FA wait: windows -> {windows}"))
            last_windows = windows

        # Opportunistic: handle an 'Existing session detected' modal if
        # one appears after our initial handle_post_login_dialogs pass.
        # It's normal for this dialog to render several seconds after
        # Log In is clicked (Gateway has to do a network round-trip to
        # discover the existing session). If we catch it here, click
        # through it and continue waiting for 2FA / API port.
        existing_session_window = None
        for type_, title, modal in windows:
            if modal and ("existing session" in title.lower()
                          or "another session" in title.lower()):
                existing_session_window = (type_, title, modal)
                break
        if existing_session_window is None and windows:
            _note_existing_session_absent()
        if existing_session_window is not None:
            log.info(f"Late existing-session dialog detected in 2FA wait: "
                     f"{existing_session_window}")
            if not handle_existing_session_dialog():
                log.error("Late existing-session dialog handling failed")
                return False
            # Re-poll windows on next iteration — the dialog is gone and
            # something new (2FA, startup parameters, etc.) may be up.
            last_windows = None
            time.sleep(0.5)
            continue

        # Look for the 2FA dialog
        two_fa_window = None
        for type_, title, modal in windows:
            if TWOFA_WINDOW_SUBSTR in title:
                two_fa_window = (type_, title, modal)
                break

        # Inspect the generic dialog before classifying it as TOTP/IB Key
        # or rejecting a separately titled passkey/browser window.
        if two_fa_window is not None:
            passkey_result = _handle_passkey_prompt(two_fa_window[1])
            if passkey_result is not None:
                return passkey_result

        # Passkey / WebAuthn: IBKR moved some accounts (Hong Kong and
        # Japan as of 2026-08, possibly wider over time) off TOTP onto
        # passkeys. Gateway then launches an in-app browser (jxbrowser)
        # that expects a hardware security key — which a headless
        # container has no way to present. Detect it and fail loudly
        # rather than waiting out the full timeout for a TOTP dialog that
        # will never appear (issue #22). We do NOT drive the ceremony:
        # emulating a WebAuthn authenticator (a FIDO2/uhid device or a
        # browser debug channel) is out of scope for this controller and
        # can't run on arm64 anyway.
        passkey_hit = _detect_passkey_flow(windows)
        if passkey_hit is not None:
            log.error(
                f"Passkey/WebAuthn login flow detected (window matched "
                f"{passkey_hit!r}). This account is on passkey 2FA, which "
                f"launches a browser expecting a hardware security key — a "
                f"headless container can't satisfy it, so unattended login "
                f"isn't possible here.")
            log.error(
                f"ALERT_2FA_FAILED mode={TRADING_MODE} "
                "reason=\"passkey/WebAuthn 2FA flow - unattended login "
                "not supported\"")
            log.error(
                "Options: (1) if the account still offers Mobile "
                "Authenticator (TOTP), make it the login method in Client "
                "Portal -> Settings -> User Settings -> Security -> Secure "
                "Login System; (2) log in attended via VNC; (3) on arm64, "
                "Gateway currently ships no arm64 jxbrowser build, so the "
                "passkey flow can't run there at all (upstream IBKR "
                "installer issue). See issue #22.")
            return False

        if two_fa_window is not None:
            if ib_key_mode:
                # IB Key push mode: the dialog appeared, IBKR sent a
                # push notification to the user's phone. We don't type
                # anything — just wait for the dialog to go away (which
                # means the user approved) or for the API port to open.
                log.info(f"IB Key 2FA dialog detected: {two_fa_window}")
                log.info("Waiting for IB Key mobile approval — "
                         "approve on your phone via the IB Key app")
                while time.monotonic() < deadline:
                    if is_api_port_open(api_port):
                        log.info("API port opened — IB Key approval succeeded")
                        return True
                    ws = agent_windows()
                    current_title = next(
                        (t for _, t, _ in ws if TWOFA_WINDOW_SUBSTR in t), None)
                    still_there = current_title is not None
                    if still_there:
                        passkey_result = _handle_passkey_prompt(current_title)
                        if passkey_result is not None:
                            return passkey_result
                    if not still_there:
                        log.info("2FA dialog dismissed — IB Key approval "
                                 "detected, proceeding")
                        return True
                    now = time.monotonic()
                    if now - last_log > 10:
                        log.info(f"  IB Key wait t+{int(now - start)}s: "
                                 "still waiting for approval...")
                        last_log = now
                    time.sleep(1.0)
                log.warning("IB Key approval wait timed out")
                # Fall through to the timeout-action dispatch below
                break
            else:
                # TOTP mode: type the code and click OK
                log.info(f"2FA dialog detected: {two_fa_window}")
                # Multi-method accounts: Gateway's 2FA dialog may first
                # show a "Select second factor device" list (pre-selected
                # to IB Key when multiple methods are enabled) instead of
                # going straight to a code-entry prompt. If the selector
                # is present, pick the device named by TWOFA_DEVICE
                # (default "Mobile Authenticator app" in TOTP mode) and
                # click OK; the dialog then switches to the code-entry
                # prompt and the flow below types the TOTP. Without this
                # step the code would be entered into a dialog still
                # expecting IB Key approval (issue #7 — the list is
                # drivable in-JVM via JLIST_SELECT even though AT-SPI
                # can't see it; same dialog IBC's SecondFactorDevice
                # handles).
                # NOTE: the "Select second factor device" heading is a
                # JTextArea, which agent_labels() (JLabel-only) misses —
                # IBC's handler finds it via findTextArea. Use the full
                # component-tree dump instead (covers JLabel,
                # JTextComponent incl. JTextArea, and AbstractButton).
                if _twofa_selector_present(agent_window(TWOFA_WINDOW_SUBSTR)):
                    log.info("2FA dialog shows device selector — selecting "
                             f"{twofa_device!r} before TOTP entry")
                    if not agent_jlist_select(TWOFA_WINDOW_SUBSTR, twofa_device):
                        log.error("JLIST_SELECT on 2FA device selector failed")
                        log.error(
                            f"ALERT_2FA_FAILED mode={TRADING_MODE} "
                            "reason=\"JLIST_SELECT on 2FA device selector failed\"")
                        if _last_jlist_response.startswith("ERR jlist_item_"):
                            # The list came back and TWOFA_DEVICE names
                            # nothing in it (or two things): a restart
                            # types the same value at the same list.
                            return _fail_2fa_needs_operator()
                        # No reply, or the window was gone — a restart
                        # may clear it.
                        return False
                    if not agent_click_in_window(TWOFA_WINDOW_SUBSTR, "OK"):
                        log.error("CLICK_IN_WIN OK on 2FA device selector failed")
                        log.error(
                            f"ALERT_2FA_FAILED mode={TRADING_MODE} "
                            "reason=\"CLICK_IN_WIN OK on 2FA device selector failed\"")
                        return False
                    # After device selection Gateway re-issues the
                    # challenge with the chosen method: the code-entry
                    # dialog appears only after a network round-trip
                    # (observed seconds, not ms). Both the selector and
                    # the code dialog are titled "Second Factor
                    # Authentication", so window titles can't tell them
                    # apart — poll for the "Enter <method> code" prompt
                    # JLabel instead (the selector has none; its heading
                    # is a JTextArea invisible to LABELS). Log window
                    # titles only when they change (mirrors the 2FA wait
                    # loop).
                    switch_prompt = None
                    last_poll_windows = None
                    for _poll in range(15):
                        time.sleep(1.0)
                        titles = [t for _, t, _ in agent_windows()]
                        if titles != last_poll_windows:
                            log.info(f"Post-select windows: {titles}")
                            last_poll_windows = titles
                        switch_prompt = _twofa_requested_method(
                            agent_labels(), window_substr=TWOFA_WINDOW_SUBSTR)
                        if switch_prompt:
                            break
                    if not switch_prompt:
                        # Issue #20: on current Gateway (observed on
                        # 10.45.1c) the in-dialog switch is rejected
                        # server-side — the first method's challenge is
                        # already in flight when the selector opens, and
                        # switching invalidates the session ("Re-login
                        # is required"). Fail with a dedicated reason
                        # rather than falling through to a misleading
                        # SETTEXT failure.
                        #
                        # Issue #37 (10.45.1j) shows the mechanism in
                        # launcher.log: selecting another method makes
                        # Gateway open a SECOND auth session ("Attempt
                        # 2: Authenticating"), and IBKR kicks the first
                        # one — "COMPETE: session kicked out" followed
                        # by "Disconnect all farms due to competing
                        # session". Hence the modal the operator sees.
                        # Report what is actually on screen rather than
                        # assuming the cause: the poll loop above already
                        # collected the window titles, and the kicked
                        # session announces itself with a "Re-login is
                        # required" modal (#37). Without it, the switch
                        # may simply not have produced a prompt yet.
                        kicked = any("re-login" in (t or "").lower()
                                     for t in (last_poll_windows or []))
                        if kicked:
                            log.error(
                                "No 2FA code-entry prompt appeared within "
                                "15s of device selection, and Gateway is "
                                "showing 'Re-login is required': the "
                                "pre-selected method's challenge is already "
                                "in flight when the selector opens, so "
                                "switching starts a second auth session and "
                                "IBKR kicks the first one (launcher.log: "
                                "'COMPETE: session kicked out'). Issues "
                                "#20, #37.")
                        else:
                            log.error(
                                "No 2FA code-entry prompt appeared within "
                                "15s of device selection. Windows on "
                                f"screen: {last_poll_windows}. On the "
                                "accounts seen so far this means IBKR "
                                "rejected the mid-login method switch; "
                                "Gateway's launcher.log shows it as "
                                "'COMPETE: session kicked out'. Issues "
                                "#20, #37.")
                        log.error(
                            f"ALERT_2FA_FAILED mode={TRADING_MODE} "
                            "reason=\"2FA device switch produced no "
                            "code-entry dialog\"")
                        log.error(
                            f"Remediation: make {twofa_device!r} the ONLY "
                            "2FA method on the account (Client Portal → "
                            "Settings → User Settings → Security → Secure "
                            "Login System). With one method Gateway shows "
                            "no selector, and the code is typed "
                            "automatically. Until then this account needs "
                            "a human: leave TWOFACTOR_CODE unset and "
                            "approve the IB Key push on your phone, or "
                            "finish the login over VNC. See "
                            "docs/UPGRADING.md (issues #7, #20, #37).")
                        if kicked:
                            return _fail_2fa_needs_operator()
                        # No "Re-login is required" modal: the code prompt
                        # may simply not have rendered within 15 s on a
                        # slow round-trip. Let the container restart
                        # rather than halting on a maybe.
                        return False
                # v0.7.0 (issue #7): on a multi-method account Gateway's
                # dialog is pre-defaulted to one method and shows an
                # "Enter <method> code" prompt. Only type our TOTP if the
                # dialog is asking for the method we can satisfy; if it
                # POSITIVELY names a different method, fail loudly with
                # remediation instead of mis-typing the code (the silent
                # failure reported in #7). Lenient: an unrecognized /
                # absent prompt falls through to the existing behavior,
                # so single-method accounts are unaffected.
                #
                # NOTE: agent_labels() filters by label TEXT, not window
                # title — so we fetch ALL labels (no filter) and scope to
                # the Second Factor window via the returned window-title.
                # Poll briefly in case the prompt JLabel renders a beat
                # after the dialog window appears.
                prompt = None
                for _ in range(3):
                    prompt = _twofa_requested_method(
                        agent_labels(), window_substr=TWOFA_WINDOW_SUBSTR)
                    if prompt:
                        break
                    time.sleep(0.5)
                if _twofa_method_mismatch(prompt, twofa_device):
                    log.error(
                        f"2FA dialog is requesting a different method than "
                        f"configured: prompt={prompt!r}, expected "
                        f"{twofa_device!r}. NOT entering the TOTP code into "
                        f"the wrong method.")
                    log.error(
                        f"ALERT_2FA_FAILED mode={TRADING_MODE} "
                        f"reason=\"2FA method mismatch\" "
                        f"dialog_prompt={prompt!r} expected={twofa_device!r}")
                    log.error(
                        "Remediation: this account has multiple 2FA methods "
                        "and Gateway defaulted to one ibg-controller can't "
                        "drive automatically yet. Set your IBKR account's "
                        f"preferred 2FA method to {twofa_device!r} (the one "
                        "matching TWOFACTOR_CODE). See docs/UPGRADING.md "
                        "(issue #7).")
                    return _fail_2fa_needs_operator()
                if prompt:
                    log.info(f"2FA method prompt {prompt!r} matches "
                             f"{twofa_device!r}")
                code = generate_totp(TOTP_SECRET)
                log.info(f"Typing TOTP code into the 2FA dialog")
                if not agent_settext_in_window(TWOFA_WINDOW_SUBSTR, code):
                    log.error("SETTEXT_IN_WIN on 2FA dialog failed")
                    log.error(
                        f"ALERT_2FA_FAILED mode={TRADING_MODE} "
                        "reason=\"agent SETTEXT_IN_WIN on 2FA dialog failed\"")
                    return False
                time.sleep(0.5)
                log.info("Clicking OK in 2FA dialog")
                if not agent_click_in_window(TWOFA_WINDOW_SUBSTR, "OK"):
                    log.error("CLICK_IN_WIN OK on 2FA dialog failed")
                    log.error(
                        f"ALERT_2FA_FAILED mode={TRADING_MODE} "
                        "reason=\"agent CLICK_IN_WIN OK on 2FA dialog failed\"")
                    return False
                log.info("2FA handled successfully")
                _reset_ccp_backoff()
                return True

        # Periodic status line
        now = time.monotonic()
        if now - last_log > 10:
            log.info(f"  2FA wait t+{int(now - start)}s: still waiting")
            last_log = now

        # 1s instead of 500ms — halves the agent polling rate without
        # hurting perceived latency (the dialog appears on the order
        # of seconds, not hundreds of ms).
        time.sleep(1.0)

    log.warning(f"No 2FA dialog appeared within {wait_seconds}s")

    # If Gateway's login dialog is stuck in its internal "connecting to
    # server (trying for another N seconds)" retry loop, IBKR's auth
    # server isn't accepting sessions for this account right now. The
    # 2FA dialog never appeared because the auth protocol never got
    # off the ground. Re-clicking Log In here or immediately falling
    # through to TWOFA_TIMEOUT_ACTION=restart (which re-launches the
    # JVM and re-clicks) extends the lockout. Apply the same CCP
    # exponential backoff the pre-auth path uses before the relogin
    # or restart dispatch kicks in.
    if _detect_login_stuck_connecting():
        log.warning("Login dialog stuck in 'connecting to server' retry "
                    "loop — IBKR auth server isn't accepting sessions "
                    "right now. Applying CCP backoff before retry.")
        _apply_ccp_backoff()
    elif TOTP_SECRET:
        # We were configured for TOTP but no Second Factor dialog ever
        # appeared and the login isn't stuck connecting. The positive
        # passkey-window check in the loop didn't fire either (its
        # jxbrowser window title may be generic), so this is a hint, not
        # a verdict: a common cause on current accounts is that the
        # account is no longer on TOTP — moved to passkey/WebAuthn (a
        # browser flow we can't drive) or IB Key push. See issue #22.
        log.warning(
            "No Second Factor (TOTP) dialog appeared despite TWOFACTOR_CODE "
            "being set. If this account was recently moved to passkey/"
            "WebAuthn 2FA (IBKR is forcing this in some regions) or IB Key "
            "push, unattended TOTP login isn't possible — see issue #22.")

    # Optional relogin attempt before dispatching the timeout action.
    # Mirrors IBC's RELOGIN_AFTER_TWOFA_TIMEOUT behavior: if set, we
    # re-drive handle_login once on the assumption that the login form
    # is still up and the user got distracted/approval-expired.
    if relogin_after_timeout is True:
        log.info("RELOGIN_AFTER_TWOFA_TIMEOUT=yes — in-JVM relogin")
        fresh_app = find_app(APP_NAME_CANDIDATES, timeout=10,
                             match_pid=JVM_PID) or app
        # v0.4.1: route through attempt_inplace_relogin so the
        # "In-JVM relogin attempt..." log line fires for observability
        # and the dismiss-error-modal / skip-progress-dialog guards run.
        if attempt_inplace_relogin(fresh_app):
            # Re-enter the wait with a fresh timer
            log.info("Relogin triggered, waiting again for 2FA dialog")
            start = time.monotonic()
            deadline = start + wait_seconds
            last_windows = None
            while time.monotonic() < deadline:
                if is_api_port_open(api_port):
                    return True
                windows = agent_windows()
                for type_, title, modal in windows:
                    if TWOFA_WINDOW_SUBSTR in title:
                        passkey_result = _handle_passkey_prompt(title)
                        if passkey_result is not None:
                            return passkey_result
                        if not TOTP_SECRET:
                            continue
                        code = generate_totp(TOTP_SECRET)
                        if agent_settext_in_window(TWOFA_WINDOW_SUBSTR, code):
                            time.sleep(0.5)
                            if agent_click_in_window(TWOFA_WINDOW_SUBSTR, "OK"):
                                log.info("2FA handled successfully (post-relogin)")
                                _reset_ccp_backoff()
                                return True
                time.sleep(1.0)
            log.warning("2FA dialog didn't appear in relogin window either")

    # Dispatch the configured timeout action.
    if timeout_action == "exit":
        log.error("TWOFA_TIMEOUT_ACTION=exit — controller exiting")
        log.error(
            f"ALERT_2FA_FAILED mode={TRADING_MODE} "
            "reason=\"2FA dialog timeout; TWOFA_TIMEOUT_ACTION=exit\"")
        try:
            os.unlink(READY_FILE)
        except FileNotFoundError:
            pass
        sys.exit(3)
    elif timeout_action == "restart":
        log.warning("TWOFA_TIMEOUT_ACTION=restart — invoking do_restart_in_place")
        if do_restart_in_place():
            return True
        log.error("Restart after 2FA timeout failed — exiting")
        log.error(
            f"ALERT_2FA_FAILED mode={TRADING_MODE} "
            "reason=\"2FA dialog timeout and do_restart_in_place failed\"")
        sys.exit(4)
    else:
        # 'none' or unrecognized — fall through to wait_for_api_port so
        # the paper-account / autorestart-token path (no 2FA) still
        # works. This preserves Phase 1 behavior.
        log.info(f"TWOFA_TIMEOUT_ACTION={timeout_action!r} — falling through "
                 "to wait_for_api_port")
        return True


def dismiss_post_login_disclaimers(timeout=30):
    """Click through known post-login disclaimer dialogs by their button name.

    These are informational dialogs Gateway shows after login (paper-trading
    risk disclaimer, terms-of-service updates, etc.) that IBC normally
    auto-clicks via its BypassWarning-style settings.

    Iterates SAFE_DISMISS_BUTTONS in preferred order (built-in defaults
    first, then any BYPASS_WARNING extensions). This is the same
    allowlist the opportunistic sweep in wait_for_api_port uses — keeps
    BYPASS_WARNING semantics consistent across every dismissal path the
    controller runs.

    We're conservative about which buttons we'll click — only ones that are
    UNAMBIGUOUSLY safe-to-dismiss disclaimer buttons. We deliberately do
    NOT click bare 'OK' because earlier testing showed that clicking OK on
    the 'Gateway' connection-progress dialog actually CANCELS the login.
    _resolve_safe_dismiss_buttons refuses bare 'OK' even if BYPASS_WARNING
    names it.
    """
    log.info("Looking for post-login disclaimer dialogs to dismiss")
    deadline = time.monotonic() + timeout
    dismissed = 0
    while time.monotonic() < deadline:
        _, buttons = agent_list()
        target = None
        for btn in SAFE_DISMISS_BUTTONS:
            if btn in buttons:
                target = btn
                break
        if target is None:
            if dismissed > 0:
                log.info(f"No more disclaimer buttons after dismissing {dismissed}")
            else:
                log.info("No disclaimer dialogs found")
            return
        log.info(f"Clicking disclaimer button {target!r}")
        agent_click(target)
        dismissed += 1
        time.sleep(1.5)  # let the dialog dismiss; another may appear


_DEFAULT_SAFE_DISMISS_BUTTONS = (
    "I understand and accept",
    "I Understand And Accept",
    "I Accept",
    "Acknowledge",
    "Accept and Continue",
)


def _resolve_safe_dismiss_buttons():
    """IBC-compat: let users extend the disclaimer allowlist via env.

    BYPASS_WARNING (comma-separated button labels) appends entries to
    the allowlist after the built-in defaults, in user-specified order.
    Consumed by both dismiss_post_login_disclaimers() and the
    opportunistic sweep inside wait_for_api_port() — so BYPASS_WARNING
    takes effect everywhere the controller dismisses disclaimers, not
    just one of those paths. Users migrating from IBC who relied on
    IBC's BypassWarning / BypassNoSecurityDialog behaviour can
    reproduce it by setting BYPASS_WARNING to the exact button text
    they want auto-dismissed.

    Returns an ordered tuple so click-preference is deterministic: the
    built-in set first (most common disclaimer labels), then
    user-added entries in the order they appear in the env var.

    Safety: we still refuse to dismiss bare "OK" even if the user
    names it — clicking OK on Gateway's "Connecting to server..."
    modal cancels the in-progress login (see ARCHITECTURE.md). That
    dialog is unrecognizable from an informational popup, so "OK" is
    permanently on the deny list.
    """
    ordered = list(_DEFAULT_SAFE_DISMISS_BUTTONS)
    seen = set(ordered)
    extra_raw = os.environ.get("BYPASS_WARNING", "").strip()
    if extra_raw:
        # Allow comma-separated OR semicolon-separated for convenience
        parts = [p.strip() for p in extra_raw.replace(";", ",").split(",")]
        added = []
        for p in parts:
            if not p:
                continue
            if p.lower() == "ok":
                log.warning("BYPASS_WARNING: refusing to add bare 'OK' to "
                            "the dismiss allowlist — clicking OK on "
                            "Gateway's 'Connecting to server...' modal "
                            "cancels the login. Use the exact dialog "
                            "button text instead.")
                continue
            if p in seen:
                continue
            ordered.append(p)
            seen.add(p)
            added.append(p)
        if added:
            log.info(f"BYPASS_WARNING: extended dismiss allowlist with {added}")
    return tuple(ordered)


SAFE_DISMISS_BUTTONS = _resolve_safe_dismiss_buttons()


def api_port_for_mode():
    """Return the API listen port for the current trading mode.

    run.sh exports API_PORT per mode and product (Gateway 4001 live /
    4002 paper, TWS 7496 / 7497); without it, the product's defaults.
    This used to assume Gateway's ports, so with TWS every readiness and
    health check probed a port TWS doesn't listen on (issue #58).
    """
    raw = os.environ.get("API_PORT", "").strip()
    if raw.isdigit() and 0 < int(raw) < 65536:
        return int(raw)
    if GATEWAY_OR_TWS == "tws":
        return 7497 if TRADING_MODE == "paper" else 7496
    return 4002 if TRADING_MODE == "paper" else 4001


# --- Incoming API connections (issue #58) ---------------------------------
# TWS and Gateway ask "Accept incoming connection attempt from <ip>?" when
# an API client connects from an address not in Trusted IPs. IBC answered
# it through AcceptIncomingConnectionAction, which gnzsnz images expose as
# TWS_ACCEPT_INCOMING; the controller ignored it, so the dialog waited for
# a click while the client hung. Same recognition and buttons as IBC: a
# label containing "Accept incoming connection"; accept clicks OK or Yes,
# reject clicks No; manual (the default) leaves it alone. Adding the
# client to Trusted IPs is safer and avoids the dialog entirely.
_INCOMING_LABEL = "Accept incoming connection"
_incoming_warned = set()
_incoming_last_windows = None


def _incoming_connection_action():
    """'accept', 'reject', or None (manual, unset, or not understood)."""
    v = os.environ.get("TWS_ACCEPT_INCOMING", "").strip().lower()
    if v == "accept" and "accept" not in _incoming_warned:
        _incoming_warned.add("accept")
        log.warning("TWS_ACCEPT_INCOMING=accept: API connections from "
                    "addresses outside Trusted IPs will be accepted. Adding "
                    "your clients to Trusted IPs instead is safer.")
    if v in ("accept", "reject"):
        return v
    if v not in ("", "manual") and "invalid" not in _incoming_warned:
        _incoming_warned.add("invalid")
        log.warning(f"TWS_ACCEPT_INCOMING={v!r} not understood; use accept, "
                    "reject or manual. Leaving incoming-connection dialogs "
                    "to you.")
    return None


def _handle_incoming_connection_dialog():
    """Answer one incoming-connection dialog per TWS_ACCEPT_INCOMING.
    True when it clicked. Reads labels only when the window list changed,
    so the per-second check costs one cheap agent call."""
    global _incoming_last_windows
    action = _incoming_connection_action()
    if action is None:
        return False
    try:
        windows = agent_windows()
    except Exception:
        return False
    key = tuple(sorted((title, modal) for _w, title, modal in windows))
    if key == _incoming_last_windows:
        return False
    try:
        hits = agent_labels(_INCOMING_LABEL)
    except Exception:
        return False
    if not hits:
        _incoming_last_windows = key
        return False
    title, text = hits[0]
    # CLICK_IN_WIN clicks the first matching button in any window whose
    # title contains this one, modal dialogs first. With two dialogs
    # matching, "Yes" could land on the other one (a confirmation, say),
    # so refuse rather than guess.
    # Exactly one window, of any kind, may match: otherwise the "OK" pass
    # can fall through to another window with that title (the main frame,
    # or another dialog) before "Yes" is tried. An empty title matches
    # every window.
    if not title or sum(1 for _w, t, _m in windows if title in t) != 1:
        log.warning(f"Incoming-connection dialog found, but its title "
                    f"{_redact_logs(title)!r} doesn't name exactly one "
                    "window; leaving it for you rather than risk clicking "
                    "the wrong one")
        _incoming_last_windows = key
        return False
    buttons = ("OK", "Yes") if action == "accept" else ("No",)
    for button in buttons:
        if agent_click_in_window(title, button):
            log.info(f"Incoming API connection "
                     f"{'accepted' if action == 'accept' else 'rejected'} "
                     f"(TWS_ACCEPT_INCOMING={action}): {_redact_logs(text)!r}")
            _incoming_last_windows = None  # another may follow at once
            return True
    log.warning(f"Incoming-connection dialog found but no "
                f"{' or '.join(buttons)} button to click in "
                f"{_redact_logs(title)!r}; leaving it for you")
    _incoming_last_windows = key
    return False


def _sleep_watching_incoming(seconds):
    """The monitor loop's pause. With TWS_ACCEPT_INCOMING set it checks for
    the incoming-connection dialog every second, because the client waits
    on it; otherwise a plain sleep."""
    if _incoming_connection_action() is None:
        time.sleep(seconds)
        return
    for _ in range(int(seconds)):
        _handle_incoming_connection_dialog()
        time.sleep(1)


def is_api_port_open(port=None):
    """Quick non-blocking probe of Gateway's API port. Returns True if a TCP
    connection succeeds. Used both for the initial readiness wait and for
    the long-running monitor loop's heartbeat."""
    if port is None:
        port = api_port_for_mode()
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(0.5)
    try:
        s.connect(("127.0.0.1", port))
        return True
    except (ConnectionRefusedError, socket.timeout, OSError):
        return False
    finally:
        try:
            s.close()
        except Exception:
            pass


# ── Post-login configuration (Phase 2.2) ──────────────────────────────
#
# Honor IBC's post-login config env vars by driving Gateway's
# Configure → Settings dialog. Runs after wait_for_api_port so we know
# the config tree has live data. Non-fatal on failure.

CONFIG_WINDOW_TITLE_SUBSTR = "Trader Workstation Configuration"


def _config_open():
    """Open the Configure → Settings dialog. Idempotent — returns True
    whether it's newly opened or was already open. Returns False if the
    dialog couldn't be opened (e.g. main window not in a state to
    accept menu input).

    v0.6.2: hardened against a dual-mode-only race that surfaced after
    v0.6.1's diagnostic logging revealed it. Live mode transitions
    API_WAIT → CONFIG within ~3 s of the API port opening, with the
    EDT still processing post-2FA tear-down (modal=true 2FA dialog
    being disposed, "Authenticating..." overlay being torn down).
    Paper mode, which skips 2FA, hits the same code path with a
    quiescent EDT.

    Symptoms (pre-v0.6.2): agent CLICK 'Configure' returns OK (the
    JMenu's selected property flips), but the JPopupMenu's heavyweight
    peer window hasn't been realized by the time the follow-up
    CLICK 'Settings' walks Window.getWindows(). The agent reports
    ``ERR not_found type=button name=Settings`` and post-login config
    silently fails on live but works on paper.

    Three changes:

    1. Wait for the post-login window stack to settle before clicking
       Configure — no modal dialogs visible, no "Authenticating..."
       window present. Bounded so we eventually proceed even on
       Gateway showing a transitional window we haven't explicitly
       named.
    2. Inter-click delay between Configure and Settings raised from
       0.3 s to 1.0 s. That's the time the EDT needs to realize the
       JPopupMenu's heavyweight peer window when it's coming out of
       a busy state. Empirically 0.3 s was enough on a quiescent EDT
       (paper) and not enough on a busy one (live post-2FA).
    3. Outer retry: up to 3 attempts to open the dialog, with a 1 s
       gap between attempts. Backstop for any other transient.
    """
    # Check if it's already open
    windows = agent_windows() or []
    if any(CONFIG_WINDOW_TITLE_SUBSTR in title for _, title, _ in windows):
        return True

    # Wait for the post-login window stack to settle. Targets the live
    # post-2FA transitional state that lingers a second or two past
    # API-port-open. Returns immediately if already settled.
    settle_deadline = time.monotonic() + 5.0
    while time.monotonic() < settle_deadline:
        windows = agent_windows() or []
        modal_present = any(modal for _, _, modal in windows)
        authenticating_present = any(
            "Authenticating" in title for _, title, _ in windows)
        if not modal_present and not authenticating_present:
            break
        time.sleep(0.3)
    else:
        log.info(
            "_config_open: window stack did not fully settle within 5 s; "
            "attempting Configure → Settings anyway")

    last_failure = "no attempt made"
    for attempt in range(1, 4):
        # Click the Configure menu. In the post-login main window this
        # is a JMenu in a JMenuBar. Clicking it via doClick() in the
        # agent flips the JMenu's selected property and the EDT
        # subsequently realizes a JPopupMenu (heavyweight window on
        # Linux Swing) holding the child JMenuItems.
        if not agent_click("Configure"):
            last_failure = "Configure click failed"
            log.info(
                f"_config_open: {last_failure} "
                f"(attempt {attempt}/3) — retrying after 1 s")
            time.sleep(1.0)
            continue
        # Let the JPopupMenu's heavyweight peer window be realized
        # before we walk Window.getWindows() looking for "Settings".
        # 1.0 s is empirically enough even on a busy EDT post-2FA.
        time.sleep(1.0)
        if not agent_click("Settings"):
            last_failure = "Settings JMenuItem not findable"
            log.info(
                f"_config_open: {last_failure} "
                f"(attempt {attempt}/3) — retrying after 1 s")
            time.sleep(1.0)
            continue
        # Wait for the Configuration dialog itself to render.
        for _ in range(20):  # up to 4 s
            windows = agent_windows() or []
            if any(CONFIG_WINDOW_TITLE_SUBSTR in title
                   for _, title, _ in windows):
                return True
            time.sleep(0.2)
        last_failure = "dialog did not render after Settings click"
        log.info(
            f"_config_open: {last_failure} "
            f"(attempt {attempt}/3) — retrying after 1 s")
        time.sleep(1.0)

    log.warning(
        f"_config_open: failed to open Configure → Settings after "
        f"3 attempts; last failure: {last_failure}")
    return False


def _config_close(action="OK"):
    """Close the config dialog by clicking OK (apply + close) or
    Cancel (discard + close). Uses CLICK_IN_WIN to scope the click to
    the dialog's own OK button and not the main window."""
    return agent_click_in_window(CONFIG_WINDOW_TITLE_SUBSTR, action)


def _coerce_yes_no(val):
    """Translate IBC-style yes/no strings to Python bool. Returns
    None for unrecognized values (so the caller can skip the knob)."""
    if val is None:
        return None
    v = str(val).strip().lower()
    if v in ("yes", "true", "1", "on"):
        return True
    if v in ("no", "false", "0", "off"):
        return False
    return None


def _lock_exit_time_visible(expected):
    """True if ``expected`` appears in the Lock and Exit panel right now.

    Reads the live component dump of the already-open config dialog, so
    it costs one agent round-trip and needs no new agent command. The
    time controls themselves sit past the dump's depth cap; what is
    visible is Gateway's summary label ("at 05:01 PM"), which showed
    "11:39 PM" after "11:39 AM" was written (field test 2026-09-27), so
    it reflects how Gateway read the value.
    Returns None when the panel can't be read at all, which the caller
    reports as "unverified" rather than "wrong".
    """
    dump = agent_window(CONFIG_WINDOW_TITLE_SUBSTR)
    if not dump or not dump.startswith("OK"):
        return None
    # An AM/PM value is compared as a time against Gateway's own summary
    # ("at 05:01 PM"), so "1:01 PM" can't pass against "11:01 PM" and
    # "5:01 PM" still matches "05:01 PM". Other values keep the substring
    # match, which survives the widget being a JSpinner editor, a
    # formatted field, or a plain JTextField.
    want = (_parse_wall_clock(expected)
            if re.search(r"[AaPp][Mm]\s*$", expected.strip()) else None)
    if want is None:
        return expected.strip() in dump
    for m in re.finditer(r"(?<!\d)(\d{1,2}:\d{2}\s*[AaPp][Mm])(?![A-Za-z])", dump):
        if _parse_wall_clock(m.group(1)) == want:
            return True
    return False


_LOGOFF_RADIO = ("Auto logoff", "Auto Logoff", "Auto log off", "Auto Log Off")
_RESTART_RADIO = ("Auto restart", "Auto Restart")


def _lock_exit_label_shown(label):
    """True when the Lock and Exit panel currently shows ``label``."""
    try:
        return any(label in text for _t, text in agent_labels(label))
    except Exception:
        return False


def _select_lock_exit_radio(*names):
    """Select the first of ``names`` that the Lock and Exit panel has as a
    radio button. IBC matches these without regard to case, the agent's
    JCHECK exactly, so callers pass the casings Gateway may use.
    Returns "changed", "unchanged" (it was already selected) or None."""
    for name in names:
        try:
            resp = _agent_request(f"JCHECK {CONFIG_WINDOW_TITLE_SUBSTR}|{name}|true")
        except Exception as e:
            log.error(f"agent JCHECK {name!r}: {type(e).__name__}: {e}")
            continue
        if resp.startswith("OK changed"):
            return "changed"
        if resp.startswith("OK"):
            return "unchanged"
    return None


def _set_meridiem(value):
    """Select Gateway's AM or PM radio button to match ``value``.

    The time field holds only the digits; AM and PM are radio buttons
    beside it, which IBC's ConfigureAutoLogoffOrRestartTimeTask sets the
    same way. Without this an AM time kept the panel's previous PM: on
    2026-09-27 writing "11:39 AM" stored 11:39 PM. Values without AM/PM
    are left alone. If the radio can't be selected, the read-back after
    OK reports the mismatch (ALERT_CONFIG_NOT_APPLIED), as before.
    """
    m = re.search(r"([AaPp][Mm])\s*$", value.strip())
    if not m:
        return False
    which = m.group(1).upper()
    if _select_lock_exit_radio(which):
        return True
    log.warning(f"  Could not select {which}; Gateway may keep its previous "
                "AM/PM, and the read-back below will say so")
    return False


def _verify_lock_exit_time_persisted(label, expected):
    """Re-open Configure → Settings after OK and confirm the Lock and Exit
    time really stuck. Returns True/False, or None if it couldn't be
    checked.

    Why re-open rather than trust the write: on 2026-09-07 a production
    box logged "Setting Auto Log Off Time = 05:01 PM" followed by
    "Post-login config applied and dialog closed", and then ran 26 hours
    straight through the configured boundary without ever logging off.
    ``agent_settext_by_label`` reported success, so the in-dialog write
    is not evidence that Gateway accepted or kept the value. Only a
    read-back after the commit is.

    Failure here is never fatal: the session is already logged in and
    usable, so the worst case is that we say the setting is unverified
    instead of silently claiming it was applied.
    """
    try:
        if not _config_open():
            log.warning("  Could not re-open config dialog to verify "
                        f"{label!r}; treating as unverified")
            return None
        try:
            if not agent_jtree_select_path(CONFIG_WINDOW_TITLE_SUBSTR,
                                           "Lock and Exit"):
                log.warning("  Could not re-select Lock and Exit to verify "
                            f"{label!r}; treating as unverified")
                return None
            time.sleep(1.0)  # same panel-render settle as the write path
            return _lock_exit_time_visible(expected)
        finally:
            # Cancel, never OK: verification must not itself commit
            # anything, and an unchanged form should leave no side effect.
            if not _config_close("Cancel"):
                log.warning("  Could not close the verification dialog with "
                            "Cancel — check for a stray config window")
    except Exception as e:
        log.warning(f"  Verification of {label!r} raised "
                    f"{type(e).__name__}: {e}; treating as unverified")
        return None


def _verify_checkbox_persisted(page, label, expected):
    """Re-open Configure → Settings after OK and confirm the checkbox
    ``label`` on ``page`` holds ``expected``. Returns True/False, or None
    if it couldn't be checked.

    A click that succeeded in the dialog doesn't show that Gateway kept
    the change; a read-back after the commit does.
    """
    try:
        if not _config_open():
            log.warning("  Could not re-open config dialog to verify "
                        f"{label}; treating as unverified")
            return None
        try:
            if not agent_jtree_select_path(CONFIG_WINDOW_TITLE_SUBSTR, page):
                log.warning(f"  Could not re-select {page} to verify "
                            f"{label}; treating as unverified")
                return None
            time.sleep(0.5)  # same panel-render settle as the write path
            state = agent_jstate(CONFIG_WINDOW_TITLE_SUBSTR, label)
            return None if state is None else state == expected
        finally:
            # Cancel, never OK: verification must not itself commit anything.
            if not _config_close("Cancel"):
                log.warning("  Could not close the verification dialog with "
                            "Cancel — check for a stray config window")
    except Exception as e:
        log.warning(f"  Verification of {label} raised "
                    f"{type(e).__name__}: {e}; treating as unverified")
        return None


def _verify_read_only_api_persisted(expected):
    """Read Read-Only API back after OK. While the box is ticked the API
    answers trading requests with error 321 (#69)."""
    return _verify_checkbox_persisted("API/Settings", "Read-Only API", expected)


# Gateway labels the API → Precautions box one of two ways, depending on
# the account (10.50.1f, feature.configure.plugins.custom.panel.R).
_BYPASS_PRECAUTIONS_LABELS = (
    "Bypass Order Precautions for API Orders",
    "Bypass Order and Ticket Precautions for API Orders and Tickets",
)


def _bypass_precautions_label():
    """(label, None) for the bypass box on the open API → Precautions page,
    or (None, reason) if neither label is showing and enabled. JSTATE only
    reads, and asking directly keeps the label that isn't there out of the
    log."""
    ambiguous = False
    for label in _BYPASS_PRECAUTIONS_LABELS:
        try:
            resp = _agent_request(f"JSTATE {CONFIG_WINDOW_TITLE_SUBSTR}|{label}") or ""
        except Exception as e:
            log.warning(f"agent JSTATE {label!r}: {type(e).__name__}: {e}")
            return None, "the agent did not answer"
        if resp.startswith("OK value="):
            return label, None
        ambiguous = ambiguous or "jcheck_ambiguous" in resp
    if ambiguous:
        return None, "more than one box on API → Precautions has that label"
    return None, "the bypass box is not showing, or is disabled, on API → Precautions"


# Gateway 10.50.1f answers OK with this notice when the commit newly
# selects Auto restart: a modal titled "IBKR Gateway", like the main
# frame, reading "You have elected to have your trading platform restart
# automatically on a daily basis…" with one OK button (monitored_panels.p).
# OK waits on it: until it's answered the dialog stays open and the rest
# of the commit, Read-Only API included, isn't saved (#69). The title is
# shared, so the notice is recognised by its text.
_AUTO_RESTART_NOTICE_TEXT = "restart automatically on a daily basis"
_CONFIG_COMMIT_TIMEOUT = 15.0


def _modal_texts(windows):
    """{title: [text lines]} for each modal dialog in an agent_windows()
    list, read from its WINDOW dump."""
    out = {}
    for title in dict.fromkeys(t for _w, t, modal in windows if modal):
        out[title] = [ln.strip() for ln in _modal_sections(agent_window(title))
                      if ' text="' in ln]
    return out


def _wait_config_committed(timeout=None):
    """After OK, wait for the settings dialog to close, acknowledging
    Gateway's Automatic Restart notice if it comes up.

    Returns True once the dialog is gone, False if it's still open at the
    deadline (the open dialogs and their text are logged), or None if the
    window list couldn't be read at all.
    """
    timeout = _CONFIG_COMMIT_TIMEOUT if timeout is None else timeout
    # Bounded by both: polls keep it short when sleep is instant, the clock
    # when each agent call is slow.
    polls = max(1, int(timeout / 0.5))
    deadline = time.monotonic() + timeout
    last = []
    next_check = 0
    acknowledged = False
    for i in range(polls):
        if i and time.monotonic() >= deadline:
            break
        windows = agent_windows() or []
        if windows:
            last = windows
            if not any(CONFIG_WINDOW_TITLE_SUBSTR in t for _w, t, _m in windows):
                return True
            if i >= next_check and any(m for _w, _t, m in windows):
                # CLICK_IN_WIN only queues the click; look again in 3 s.
                next_check = i + 6
                modals = _modal_texts(windows)
                notice = next((title for title, lines in modals.items()
                               if any(_AUTO_RESTART_NOTICE_TEXT in ln for ln in lines)),
                              None)
                # CLICK_IN_WIN clicks the first OK in any modal whose title
                # contains this one, so click only when the notice is the
                # sole such modal; anything else stays up for the operator.
                if notice and sum(1 for _w, t, m in windows if m and notice in t) == 1:
                    if agent_click_in_window(notice, "OK") and not acknowledged:
                        log.info("  Acknowledged Gateway's Automatic Restart notice")
                        acknowledged = True
        time.sleep(0.5)
    if not last:
        return None
    modals = _modal_texts(last)
    log.warning(_redact_logs(
        f"  Settings dialog still open after OK; dialogs showing: "
        f"{', '.join(modals) or 'none'}"))
    for title, lines in modals.items():
        for ln in lines[:5]:
            log.warning(_redact_logs(f"    [{title}] {ln}"))
    return False


def handle_post_login_config():
    """Drive Gateway's Configure → Settings dialog to apply env-var
    overrides for the IBC-compatible API config knobs.

    Env vars (matching IBC's names where possible):

      Present in Gateway's API → Settings panel (always):
        TWS_MASTER_CLIENT_ID — integer, API → Settings → Master client ID
        READ_ONLY_API        — yes/no, API → Settings → Read-Only API

      Present in Gateway's API → Precautions panel:
        BYPASS_ORDER_PRECAUTIONS — yes/no, "Bypass Order Precautions for
                               API Orders" (or Gateway's longer "...and
                               Ticket Precautions..." label)

      Present in Gateway's Lock and Exit panel (one of two, depending
      on whether the account has the autorestart daily-token cycle
      enabled — Gateway shows *either* Auto Log Off Time *or* Auto
      Restart Time but not both):
        AUTO_LOGOFF_TIME  — HH:MM, "Set Auto Log Off Time (HH:MM)"
        AUTO_RESTART_TIME — HH:MM AM/PM, "Set Auto Restart Time (HH:MM)"
      The handler tries both labels and sets the one Gateway is
      currently displaying. If the user set the one that Gateway
      isn't showing in this session, a clear warning is logged.

      NOT present in Gateway's config dialog at all (TWS-only; we warn):
        ALLOW_BLIND_TRADING  — TWS Precautions tab; Gateway's
                               Precautions page has no such box
        SAVE_TWS_SETTINGS    — not a Gateway knob

    Skips any knob whose env var isn't set or is empty. Non-fatal on
    per-knob failure — logs a warning and moves on so users still get
    a working API port even if one setting couldn't be applied.
    """
    global _logoff_backstop_armed
    master_client_id = os.environ.get("TWS_MASTER_CLIENT_ID", "").strip()
    read_only_api_raw = os.environ.get("READ_ONLY_API", "")
    read_only_api = _coerce_yes_no(read_only_api_raw)
    auto_logoff_time = os.environ.get("AUTO_LOGOFF_TIME", "").strip()
    auto_restart_time = os.environ.get("AUTO_RESTART_TIME", "").strip()
    bypass_precautions_raw = os.environ.get("BYPASS_ORDER_PRECAUTIONS", "")
    bypass_precautions = _coerce_yes_no(bypass_precautions_raw)
    # Truly TWS-only — not present in any Gateway config state we've observed
    allow_blind_trading = _coerce_yes_no(os.environ.get("ALLOW_BLIND_TRADING", ""))
    save_tws_settings = os.environ.get("SAVE_TWS_SETTINGS", "").strip()

    wanted_api_tab = master_client_id or read_only_api is not None
    wanted_precautions_tab = bypass_precautions is not None
    wanted_lock_exit_tab = bool(auto_logoff_time) or bool(auto_restart_time)
    wanted_anything = wanted_api_tab or wanted_precautions_tab or wanted_lock_exit_tab

    # Early warning for the truly TWS-only vars
    tws_only_set = []
    if allow_blind_trading is not None:
        tws_only_set.append("ALLOW_BLIND_TRADING")
    if save_tws_settings:
        tws_only_set.append("SAVE_TWS_SETTINGS")
    if tws_only_set:
        log.warning("Post-login config: these env vars are not present in "
                    f"Gateway's config dialog and are being ignored: "
                    f"{', '.join(tws_only_set)}. They're TWS-specific. "
                    "See controller/docs/MIGRATION.md for details.")

    # v0.6.1: dump exactly what we observed in os.environ on both the
    # apply and the skip paths. Pre-v0.6.1 a "no supported env vars
    # set, skipping" line gave no clue whether the user had genuinely
    # set nothing or whether the env transmission chain (Docker
    # --env-file → run.sh → start_controller → python3 fork) had
    # dropped a value somewhere along the way. The values logged here
    # are not secrets — IBC-equivalent post-login config knobs only.
    log.info(
        f"Post-login config env: TWS_MASTER_CLIENT_ID={master_client_id!r}, "
        f"READ_ONLY_API={read_only_api_raw!r} (coerced={read_only_api!r}), "
        f"AUTO_LOGOFF_TIME={auto_logoff_time!r}, "
        f"AUTO_RESTART_TIME={auto_restart_time!r}, "
        f"BYPASS_ORDER_PRECAUTIONS={bypass_precautions_raw!r} "
        f"(coerced={bypass_precautions!r})"
    )
    for name, raw, coerced in (
            ("READ_ONLY_API", read_only_api_raw, read_only_api),
            ("BYPASS_ORDER_PRECAUTIONS", bypass_precautions_raw, bypass_precautions)):
        if raw.strip() and coerced is None:
            log.warning(f"Post-login config: {name}={raw.strip()!r} isn't one of "
                        "yes/no, true/false, on/off, 1/0; leaving that setting "
                        "as Gateway has it")

    if not wanted_anything:
        log.info("Post-login config: no supported env vars set, skipping")
        return True

    log.info("Applying post-login configuration from env vars")
    if not _config_open():
        log.warning("Post-login config: could not open Configure → Settings "
                    "dialog — settings not applied")
        return False

    changed = False
    # Read back after OK, like the Lock and Exit time (#69).
    read_only_written = False
    client_id_written = False
    # The bypass box's label as Gateway showed it, once written.
    precautions_written = None
    # (label, value, env var) of the Lock and Exit time we wrote, if any.
    # Verified after OK — see _verify_lock_exit_time_persisted.
    lock_exit_written = None
    # Set when a mode switch couldn't be undone: then nothing is saved.
    abort_config = False

    if wanted_api_tab:
        log.info("  Navigating to API → Settings")
        if not agent_jtree_select_path(CONFIG_WINDOW_TITLE_SUBSTR, "API/Settings"):
            log.warning("  Could not select API/Settings tree node")
        else:
            time.sleep(0.5)  # let the right panel render
            if master_client_id:
                log.info(f"  Setting Master API client ID = {master_client_id}")
                if agent_settext_by_label(CONFIG_WINDOW_TITLE_SUBSTR,
                                          "Master API client ID",
                                          master_client_id):
                    changed = True
                    client_id_written = True
                else:
                    log.warning("  Failed to set Master API client ID")
            if read_only_api is not None:
                log.info(f"  Setting Read-Only API = {read_only_api}")
                if agent_jcheck(CONFIG_WINDOW_TITLE_SUBSTR,
                                "Read-Only API", read_only_api):
                    changed = True
                    read_only_written = True
                else:
                    log.warning("  Failed to toggle Read-Only API")

    if wanted_precautions_tab:
        log.info("  Navigating to API → Precautions")
        not_set = label = None
        if not agent_jtree_select_path(CONFIG_WINDOW_TITLE_SUBSTR, "API/Precautions"):
            not_set = "could not select API → Precautions"
        else:
            time.sleep(0.5)  # let the right panel render
            label, not_set = _bypass_precautions_label()
            if label is not None:
                log.info(f"  Setting {label} = {bypass_precautions}")
                if agent_jcheck(CONFIG_WINDOW_TITLE_SUBSTR, label, bypass_precautions):
                    changed = True
                    precautions_written = label
                else:
                    not_set = "the bypass box could not be set"
        if not_set:
            log.error(
                f"ALERT_CONFIG_NOT_APPLIED mode={TRADING_MODE} "
                f"setting=\"{label or _BYPASS_PRECAUTIONS_LABELS[0]}\" "
                f"env_var=BYPASS_ORDER_PRECAUTIONS "
                f"requested=\"{bypass_precautions_raw.strip()}\" "
                f"reason=\"{not_set}\"")

    if wanted_lock_exit_tab:
        log.info("  Navigating to Lock and Exit")
        if not agent_jtree_select_path(CONFIG_WINDOW_TITLE_SUBSTR, "Lock and Exit"):
            log.warning("  Could not select Lock and Exit tree node")
        else:
            # Gateway shows ONE of these two labels, depending on which of
            # the panel's radio buttons is selected, "Auto logoff" (the
            # default, at 11:45 PM) or "Auto restart":
            #   "Set Auto Log Off Time (HH:MM)"
            #   "Set Auto Restart Time (HH:MM)"
            # With exactly one of the two times set, select its mode first
            # (issue #65; IBC's ConfigureAutoLogoffOrRestartTimeTask does
            # the same). With both set, whichever field is displayed is
            # the one set, as before.
            time.sleep(1.0)  # panel render
            logoff_label = "Set Auto Log Off Time (HH:MM)"
            restart_label = "Set Auto Restart Time (HH:MM)"
            # With exactly one of the two times set, the env decides the
            # mode. Touch the radios only when the field we need isn't
            # showing, so a panel already in the right mode is left alone.
            want = None
            if auto_restart_time and not auto_logoff_time:
                want = (restart_label, _RESTART_RADIO, _LOGOFF_RADIO)
            elif auto_logoff_time and not auto_restart_time:
                want = (logoff_label, _LOGOFF_RADIO, _RESTART_RADIO)
            switched_back_to = None
            if want and not _lock_exit_label_shown(want[0]):
                selected = _select_lock_exit_radio(*want[1])
                if selected:
                    # Only a real change needs undoing; "unchanged" means the
                    # label check missed a mode that was already right.
                    switched_back_to = want[2] if selected == "changed" else None
                    log.info(f"  Selected {want[1][0]} (only "
                             f"{'AUTO_RESTART_TIME' if want[0] == restart_label else 'AUTO_LOGOFF_TIME'}"
                             f" is set)")
                    # The panel swaps its time label; give it a moment.
                    for _ in range(5):
                        time.sleep(1.0)
                        if _lock_exit_label_shown(want[0]):
                            break
                else:
                    log.warning(f"  Could not select {want[1][0]}; its time "
                                "can only be set while it is selected")
            for env_var, value, label, other in (
                    ("AUTO_LOGOFF_TIME", auto_logoff_time, logoff_label, "Set Auto Restart Time"),
                    ("AUTO_RESTART_TIME", auto_restart_time, restart_label, "Set Auto Log Off Time")):
                if not value:
                    continue
                log.info(f"  Setting {label.split(' (')[0][4:]} = {value}")
                if agent_settext_by_label(CONFIG_WINDOW_TITLE_SUBSTR, label, value):
                    changed = True
                    _set_meridiem(value)
                    lock_exit_written = (label, value, env_var)
                    continue
                if switched_back_to:
                    # Never save a mode with a time we didn't set: switch
                    # back and say so. If even that fails, discard the whole
                    # dialog below.
                    if _select_lock_exit_radio(*switched_back_to) != "changed":
                        abort_config = True
                    log.error(
                        f"ALERT_CONFIG_NOT_APPLIED mode={TRADING_MODE} "
                        f"setting=\"{label}\" env_var={env_var} "
                        f"requested=\"{value}\" "
                        f"reason=\"selected the matching mode but its time "
                        f"field never appeared; "
                        f"{'could not switch back, discarding the dialog' if abort_config else 'switched back'}\"")
                else:
                    log.warning(f"  Failed to set {label.split(' (')[0][4:]} — "
                                f"Gateway is showing '{other}' in this session "
                                "instead; set only the variable for the mode "
                                "you want and the controller selects it.")

    # A mode we switched and couldn't switch back must not be saved with a
    # time we didn't set: discard every change in the dialog instead.
    if abort_config:
        if not _config_close("Cancel"):
            log.warning("Could not click Cancel on the config dialog")
        for setting, env_var, value, applied in (
                ("Read-Only API", "READ_ONLY_API", read_only_api_raw,
                 read_only_api is not None),
                ("Master API client ID", "TWS_MASTER_CLIENT_ID", master_client_id,
                 bool(str(master_client_id or "").strip())),
                (precautions_written, "BYPASS_ORDER_PRECAUTIONS",
                 bypass_precautions_raw, precautions_written is not None)):
            if applied:
                log.error(
                    f"ALERT_CONFIG_NOT_APPLIED mode={TRADING_MODE} "
                    f"setting=\"{setting}\" env_var={env_var} "
                    f"requested=\"{str(value).strip()}\" "
                    f"reason=\"discarded with a Lock and Exit change that "
                    f"could not be undone\"")
        log.error("Post-login config discarded: the Lock and Exit mode was "
                  "switched and could not be switched back")
        return False

    # Apply + close the dialog (OK commits and closes).
    if changed:
        if not _config_close("OK"):
            log.warning("Could not click OK to commit config changes")
            return False
        committed = _wait_config_committed()
        if committed is False:
            # Reading back now would read the unsaved dialog, so report
            # every setting it held instead.
            unsaved = []
            if client_id_written:
                unsaved.append(("Master API client ID", "TWS_MASTER_CLIENT_ID",
                                master_client_id))
            if read_only_written:
                unsaved.append(("Read-Only API", "READ_ONLY_API", read_only_api_raw))
            if precautions_written is not None:
                unsaved.append((precautions_written, "BYPASS_ORDER_PRECAUTIONS",
                                bypass_precautions_raw))
            if lock_exit_written is not None:
                label, value, env_var = lock_exit_written
                unsaved.append((label, env_var, value))
            for setting, env_var, value in unsaved:
                log.error(
                    f"ALERT_CONFIG_NOT_APPLIED mode={TRADING_MODE} "
                    f"setting=\"{setting}\" env_var={env_var} "
                    f"requested=\"{str(value).strip()}\" "
                    f"reason=\"the settings dialog did not close after OK; "
                    f"see the dialog Gateway is showing\"")
            return False
        if committed is None:
            # The agent didn't answer for the whole wait, so the dialog may
            # still be open behind a prompt, and a read-back would read it
            # unsaved and then cancel it.
            log.warning("Could not confirm the settings dialog closed after OK; "
                        "not reading settings back. Check them under "
                        "Configure → Settings.")
            return True
        log.info("Post-login config committed; dialog closed")
        if read_only_written:
            time.sleep(1.0)  # let the dialog fully dispose before re-opening
            verified = _verify_read_only_api_persisted(read_only_api)
            if verified is True:
                log.info(f"  Verified: Read-Only API reads back as {read_only_api}")
            elif verified is False:
                # Stable grep token, as for the times below.
                log.error(
                    f"ALERT_CONFIG_NOT_APPLIED mode={TRADING_MODE} "
                    f"setting=\"Read-Only API\" env_var=READ_ONLY_API "
                    f"requested=\"{read_only_api_raw.strip()}\" "
                    f"reason=\"Gateway did not retain the value after OK; "
                    f"{'API clients can place orders' if read_only_api else 'API clients cannot place or change orders'}\"")
            else:
                log.warning("  Could not verify Read-Only API — check it "
                            "under Configure → Settings → API → Settings.")
        if precautions_written is not None:
            label = precautions_written
            time.sleep(1.0)  # let the dialog fully dispose before re-opening
            verified = _verify_checkbox_persisted("API/Precautions", label,
                                                  bypass_precautions)
            if verified is True:
                log.info(f"  Verified: {label} reads back as {bypass_precautions}")
            elif verified is False:
                log.error(
                    f"ALERT_CONFIG_NOT_APPLIED mode={TRADING_MODE} "
                    f"setting=\"{label}\" env_var=BYPASS_ORDER_PRECAUTIONS "
                    f"requested=\"{bypass_precautions_raw.strip()}\" "
                    f"reason=\"Gateway did not retain the value after OK; "
                    f"{'API orders still go through order precautions' if bypass_precautions else 'API orders still bypass order precautions'}\"")
            else:
                log.warning(f"  Could not verify {label} — check it under "
                            "Configure → Settings → API → Precautions.")
        if lock_exit_written is not None:
            label, value, env_var = lock_exit_written
            time.sleep(1.0)  # let the dialog fully dispose before re-opening
            verified = _verify_lock_exit_time_persisted(label, value)
            if verified is True:
                log.info(f"  Verified: {label} reads back as {value!r}")
                if env_var == "AUTO_LOGOFF_TIME":
                    if _gateway_zone_matches_local():
                        _logoff_backstop_armed = True
                        _note_session_login()
            elif verified is False:
                # Stable grep token: an operator who set a schedule needs
                # to know it will not happen. See docs/OBSERVABILITY.md.
                log.error(
                    f"ALERT_CONFIG_NOT_APPLIED mode={TRADING_MODE} "
                    f"setting=\"{label}\" env_var={env_var} "
                    f"requested=\"{value}\" "
                    f"reason=\"Gateway did not retain the value after OK; "
                    f"the schedule will not fire\"")
            else:
                log.warning(
                    f"  Could not verify {label} — it may or may not have "
                    "been applied. Confirm in the Gateway UI if you depend "
                    "on the schedule.")
    else:
        # Nothing was actually changed — use Cancel to avoid any
        # side-effect of OK that might otherwise fire even for an
        # unchanged form.
        if not _config_close("Cancel"):
            log.warning("Could not click Cancel on unchanged config dialog")
        log.info("Post-login config: no changes applied, dialog closed")
    return True


# ── CCP lockout detection + exponential backoff ──────────────────────
#
# IBKR's auth server silently drops fresh-password auth requests when
# it's in a "CCP lockout" state (AuthTimeoutMonitor-CCP: Timeout! in
# launcher.log, with no preceding NS_AUTH_START). The controller's
# TWOFA_TIMEOUT_ACTION=restart path was inadvertently feeding this
# lockout by immediately retrying auth with zero backoff — every retry
# extended the lockout window.
#
# The fix: after clicking Log In, poll launcher.log for the CCP Timeout
# signature. If detected, skip the 2FA wait (it can't succeed if auth
# was rejected) and enter an exponential backoff before the next retry.

_ccp_backoff_seconds = 0.0  # current backoff; 0 = no backoff active
_CCP_BACKOFF_INITIAL = 60.0
_CCP_BACKOFF_MAX = 600.0
_CCP_BACKOFF_MULTIPLIER = 2.0

# Consecutive-lockout streak for diagnosing concurrent-session lockouts.
# On 2026-04-17 live was stuck in CCP lockout for ~3 hours because another
# IBKR session (web/mobile) held the auth slot. The silent cool-downs
# couldn't clear it — only the user logging out elsewhere did. v0.4.8 adds
# streak-based messaging so future incidents are diagnosed in seconds, not
# hours: streak 2 → warn about concurrent session; streak 3+ → emit a
# structured ALERT_CCP_PERSISTENT token for external monitoring.
_ccp_lockout_streak = 0
_CCP_STREAK_WARN_CONCURRENT = 2
_CCP_STREAK_ALERT_PERSISTENT = 3

# Wall-clock timestamp of the most recent successful auth. Reported via
# the /health endpoint so monitoring can alert on "logged in at some
# point but hasn't re-authed in too long". None until the first success.
_last_auth_success_ts = None


def _detect_ccp_lockout(timeout=25):
    """Poll launcher.log for CCP auth timeout within `timeout` seconds
    of the Log In click. Returns True if CCP lockout is detected (auth
    was silently rejected), False if auth appears to have proceeded
    normally (NS_AUTH_START appeared, or no timeout within the window).

    Gateway's auth protocol:
      1. Click Log In → Gateway sends auth request
      2. Server replies with NS_AUTH_START within ~200ms (success)
         OR server is silent for 20s → CCP Timeout (lockout)

    We check by reading the LAST few lines of launcher.log repeatedly
    for up to `timeout` seconds. If we see 'AuthTimeoutMonitor-CCP:
    Timeout!' without a preceding 'NS_AUTH_START', auth was rejected.
    """
    launcher_log = os.path.join(JTS_CONFIG_DIR, "launcher.log")
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            with open(launcher_log, encoding="utf-8", errors="replace") as f:
                # Read last 4KB — enough to catch the recent auth sequence
                f.seek(0, 2)
                size = f.tell()
                f.seek(max(0, size - 4096))
                tail = f.read()
        except (FileNotFoundError, PermissionError):
            time.sleep(1)
            continue

        timeout_str = "AuthTimeoutMonitor-CCP: Timeout!"
        activate_str = "AuthTimeoutMonitor-CCP: activate"
        if timeout_str in tail:
            timeout_pos = tail.rfind(timeout_str)
            activate_pos = tail.rfind(activate_str)

            # Stale detection: if a NEW auth cycle started (activate)
            # AFTER the Timeout!, the Timeout! is from a previous
            # attempt and the current one is still in progress. Keep
            # polling — don't false-positive.
            if activate_pos > timeout_pos:
                time.sleep(1)
                continue

            # Check if NS_AUTH_START appeared AFTER the most recent
            # "Authenticating" line. If it did, auth progressed past
            # the CCP check and the timeout is a different failure.
            auth_pos = tail.rfind("Authenticating")
            ns_pos = tail.rfind("NS_AUTH_START")
            if ns_pos > auth_pos and auth_pos >= 0:
                # NS_AUTH_START came after Authenticating — auth
                # reached the server, this is a credential or
                # post-auth failure, not CCP lockout.
                return False
            log.warning("CCP LOCKOUT DETECTED — IBKR's auth server "
                        "silently dropped the auth request (no "
                        "NS_AUTH_START before Timeout)")
            global _ccp_lockout_streak
            _ccp_lockout_streak += 1
            if _ccp_lockout_streak == _CCP_STREAK_WARN_CONCURRENT:
                log.warning(
                    f"CCP lockout has hit {_ccp_lockout_streak} times in "
                    "a row. Most common cause is a concurrent IBKR "
                    "session (another TWS/Gateway) or a stranded slot "
                    f"from a prior unclean teardown on the {TRADING_MODE} "
                    "account. Remediation: log into IBKR Mobile as this "
                    "username — mobile login auto-logs-out all "
                    "TWS/Gateway sessions and is the reliable kick path. "
                    "IBKR Client Portal (web) login does NOT kick the "
                    "slot. See docs/DISCONNECT_RECOVERY.md — scenario "
                    "'CCP lockout (concurrent IBKR session)'.")
            elif _ccp_lockout_streak >= _CCP_STREAK_ALERT_PERSISTENT:
                log.error(
                    f"ALERT_CCP_PERSISTENT consecutive_lockouts="
                    f"{_ccp_lockout_streak} mode={TRADING_MODE} "
                    "suggested_action=\"log into IBKR Mobile as this "
                    "username to force-log-out the held TWS/Gateway "
                    "slot; IBKR Client Portal (web) does NOT kick the "
                    "slot\"")
            return True
        time.sleep(1)
    return False


def _apply_ccp_backoff():
    """Sleep for the current CCP backoff duration, doubling it for
    the next call. Logs the delay so the user can see what's happening.
    Returns the duration slept."""
    global _ccp_backoff_seconds
    if _ccp_backoff_seconds == 0:
        _ccp_backoff_seconds = _CCP_BACKOFF_INITIAL
    else:
        _ccp_backoff_seconds = min(
            _ccp_backoff_seconds * _CCP_BACKOFF_MULTIPLIER,
            _CCP_BACKOFF_MAX,
        )
    log.warning(f"CCP backoff: waiting {int(_ccp_backoff_seconds)}s before "
                "next auth attempt (exponential backoff to let IBKR's "
                "rate limiter clear)")
    time.sleep(_ccp_backoff_seconds)
    return _ccp_backoff_seconds


def _reset_ccp_backoff():
    """Reset the backoff + lockout streak after a successful auth. Also
    records the auth success timestamp for /health reporting."""
    global _ccp_backoff_seconds, _ccp_lockout_streak, _last_auth_success_ts
    if _ccp_backoff_seconds > 0:
        log.info("CCP backoff reset — auth succeeded")
        _ccp_backoff_seconds = 0.0
    _ccp_lockout_streak = 0
    _last_auth_success_ts = time.time()


def _detect_login_stuck_connecting():
    """Return True if the login dialog is displaying the 'connecting to
    server (trying for another N seconds)' retry-loop label.

    This is a sibling signal to _detect_ccp_lockout's launcher.log
    signature. CCP detection catches the case where IBKR's auth server
    replies with a silent `AuthTimeoutMonitor-CCP: Timeout!`; this
    helper catches an earlier failure mode where Gateway can't even
    establish the session — the login window shows an internal retry
    counter and the auth protocol never starts. Both states have the
    same remediation: back off, don't re-click Log In.

    Implementation: ask the agent for all currently-visible JLabel text
    and look for the signature substrings. Cheap (one socket round-trip,
    no polling) because the caller is already in a post-timeout
    decision path and we don't want to add another long wait on top.
    """
    try:
        labels = agent_labels()
    except Exception:
        return False
    for _wtitle, text in labels:
        lower = text.lower()
        if "connecting to server" in lower or "trying for another" in lower:
            return True
    return False


# Hard cap on in-JVM relogin retries before escalating to JVM restart.
# Reached in two situations: (a) CCP lockout keeps rearming across
# attempts, or (b) ``attempt_inplace_relogin`` returns False (disposed
# login frame — see v0.4.4). Escalation then calls
# ``_escalate_to_jvm_restart`` (v0.4.5), which long-cools-down and soft-
# restarts this mode's JVM via ``do_restart_in_place``.
_INPLACE_RELOGIN_MAX_ATTEMPTS = 8

# v0.4.5: cap on JVM-restart attempts. Each attempt does a long CCP
# cool-down + ``do_restart_in_place``. 5 × CCP_COOLDOWN_SECONDS (default
# 1200s = 20min) = 100 min of wall clock at the cap, which is more than
# enough for IBKR's CCP rate limiter to clear if it's going to clear.
# Past this cap the controller halts (since v0.13.0; it used to exit).
_JVM_RESTART_MAX_ATTEMPTS = int(os.environ.get("JVM_RESTART_MAX_ATTEMPTS", "5"))

# v0.5.9: CCP-lockout-triggered JVM restarts are now opt-in. The
# historical behaviour (5 escalations, each SIGKILL-capable on its
# teardown) was the root cause of the 2026-04-19 incident where a 24h
# retry loop re-stranded an IBKR auth slot 5 times and extended IBKR's
# server-side zombie timer each time. Default 0 means: on the first
# path that would call ``_escalate_to_jvm_restart``, emit
# ``ALERT_CCP_PERSISTENT_HALT`` and halt (since v0.13.0; it used to exit,
# which restarted the container into the same lockout) — the healthcheck
# flags the container unhealthy and the operator investigates before the
# controller re-opens the auth pipe. Set to a
# positive integer to restore the pre-v0.5.9 auto-recovery loop, capped
# at that many attempts (supersedes ``JVM_RESTART_MAX_ATTEMPTS``).
_CCP_LOCKOUT_MAX_JVM_RESTARTS = int(
    os.environ.get("CCP_LOCKOUT_MAX_JVM_RESTARTS", "0"))

# v0.4.5: long CCP cool-down before a JVM restart. Much longer than the
# exponential attempt-spacing backoff (``_apply_ccp_backoff``) — CCP
# needs silence to reset, not just spacing. Empirical observation:
# short waits (< 5 min) keep the limiter armed; 20+ min usually clears.
# Env var ``CCP_COOLDOWN_SECONDS`` (default 1200 = 20min).
_CCP_COOLDOWN_SECONDS_DEFAULT = 1200

# v0.5.5: adaptive cool-down scales with restart-attempt index. The fixed
# 1200s was sufficient for rate-limiter reset but not for stranded-session
# release — when a prior JVM's CCP session slot is still held server-side
# (see ALERT_JVM_UNCLEAN_SHUTDOWN), the next auth immediately hits lockout
# and the cycle repeats. Scaling the wait per attempt (1200 → 1800 → 2700
# → 3600 capped) gives IBKR's session-slot timeout a chance to drain even
# when our SIGTERM didn't close the socket cleanly.
# Env vars: ``CCP_COOLDOWN_MAX_SECONDS`` (default 3600),
# ``CCP_COOLDOWN_MULTIPLIER`` (default 1.5; set 1.0 to restore the
# pre-v0.5.5 fixed-duration behaviour).
_CCP_COOLDOWN_MAX_SECONDS_DEFAULT = 3600
_CCP_COOLDOWN_MULTIPLIER_DEFAULT = 1.5

# v0.5.10: IBKR runs a daily server-side maintenance window during which
# every Gateway/TWS session receives a cooperative shutdown (JVM exits
# with code 0). IBKR's system status page publishes the North America
# reset as 00:15-01:45 ET Sunday-Friday and 00:00-02:00 ET on Saturday
# (checked 2026-10-03; this comment used to say 23:45-00:15). The default
# window is 23:30-02:00 ET: it covers both, and keeps the 23:30 start
# because Gateway was seen shutting down at 23:45 ET. Accounts hosted in
# Europe or Asia reset at other times — set IBKR_RESET_WINDOWS and
# IBKR_RESET_TZ (see _reset_windows_from_env). 2026-04-20/21 production
# incident: re-auth within ~8s of the cooperative shutdown hits IBKR's
# still-draining auth server and the request is silently dropped →
# CCP LOCKOUT cascade across both live and paper. The delay lets IBKR's
# server-side session teardown propagate before we re-auth. 8 min
# default is empirical; tune via env var if a deployment sees different
# drain timing.
_RESET_WINDOWS_DEFAULT = "23:30-02:00"
_RESET_TZ_DEFAULT = "America/New_York"


def _reset_windows_from_env():
    """Parse IBKR_RESET_WINDOWS ("HH:MM-HH:MM", comma-separated for
    regions with more than one reset) and IBKR_RESET_TZ. A window whose
    end is earlier than its start crosses midnight. Anything unparseable
    falls back to the default with a warning, so a typo can never
    silently disable the guard."""
    raw = os.environ.get("IBKR_RESET_WINDOWS", "").strip() or _RESET_WINDOWS_DEFAULT
    tz = os.environ.get("IBKR_RESET_TZ", "").strip() or _RESET_TZ_DEFAULT
    try:
        ZoneInfo(tz)
        windows = []
        for part in raw.split(","):
            a, b = part.strip().split("-")
            ah, am = (int(x) for x in a.strip().split(":"))
            bh, bm = (int(x) for x in b.strip().split(":"))
            windows.append((dtime(ah, am), dtime(bh, bm)))
        if not windows:
            raise ValueError("no windows")
        return tuple(windows), tz
    except Exception as e:
        log.warning(f"IBKR_RESET_WINDOWS={raw!r} / IBKR_RESET_TZ={tz!r} not "
                    f"understood ({type(e).__name__}: {e}); using "
                    f"{_RESET_WINDOWS_DEFAULT} {_RESET_TZ_DEFAULT}")
        return ((dtime(23, 30), dtime(2, 0)),), _RESET_TZ_DEFAULT


_RESET_WINDOWS, _CCP_MAINTENANCE_WINDOW_TZ = _reset_windows_from_env()
_CCP_MAINTENANCE_RECOVERY_DELAY_SECONDS_DEFAULT = 480   # 8 min
_CCP_MAINTENANCE_RECOVERY_DELAY_SECONDS = int(os.environ.get(
    "CCP_MAINTENANCE_RECOVERY_DELAY_SECONDS",
    str(_CCP_MAINTENANCE_RECOVERY_DELAY_SECONDS_DEFAULT)))

# Issue #23: Gateway's own daily auto-restart (Configure → Lock and Exit →
# "Set Auto Restart Time", i.e. AUTO_RESTART_TIME). At that time the JVM
# spawns install4j's `.install4j/restarter`, which shuts the calling
# launcher down and re-executes it; the replacement re-uses the current
# session's credentials, so no login and no second factor are needed.
# monitor_loop saw that code-0 exit as a crash and do_restart_in_place
# launched a SECOND Gateway ~2s before install4j launched its own — two
# instances on one display and one settings dir, the agent drove the
# wrong window, login failed, the controller halted and the container
# restarted into a cold login (and a 2FA push) every night. The fix:
# when install4j's restarter.log was just written, don't launch
# anything — wait for the instance install4j is bringing up, adopt it
# (it inherits INSTALL4J_ADD_VM_PARAMS through the restarter, so its
# agent binds our socket and reports its PID) and resume monitoring.
#
# AUTO_RESTART_ADOPT=no restores the previous always-relaunch behaviour.
# Every failure inside the adoption path falls through to the
# controller-driven restart, so the worst case is the old behaviour
# plus the adopt timeout.
# PR #29 (jpike88): when Gateway's 2FA dialog shows the passkey prompt
# ("Use your Passkey device …"), the controller can press Authenticate.
# It never performs the WebAuthn ceremony itself — that must be completed
# by an authenticator you run alongside the container (a virtual FIDO
# device such as passless, a key passed through, or a person). Opt-in:
# with it unset the controller keeps v0.8.1's behaviour and fails loudly
# on a passkey prompt rather than pressing anything. Off by default
# because the maintainer has no passkey account to verify it against;
# it is validated by the contributor's live use and the flow tests.
_PASSKEY_AUTHENTICATE = _coerce_yes_no(
    os.environ.get("PASSKEY_AUTHENTICATE", "")) is True

_AUTO_RESTART_ADOPT = _coerce_yes_no(
    os.environ.get("AUTO_RESTART_ADOPT", "yes")) is not False
# How long to wait for the self-restarted JVM's agent to report a new
# PID over the inherited agent socket. Measured ~0-3s in production
# (issue #23); the default leaves room for slow hosts.
_AUTO_RESTART_ADOPT_TIMEOUT_SECONDS = int(os.environ.get(
    "AUTO_RESTART_ADOPT_TIMEOUT_SECONDS", "90"))
# restarter.log must have been written this recently for a JVM exit to
# count as a self-restart. The trigger is install4j's own file, not the
# wall-clock vs AUTO_RESTART_TIME: that comparison would have to guess
# which timezone Gateway's Lock-and-Exit field is in (jts.ini carries
# its own TimeZone, the container another) and would fire on any code-0
# exit that happened to land in the window.
_AUTO_RESTART_DETECT_WINDOW_SECONDS = 120
# Not every install4j build writes restarter.log. Verified 2026-09-06 on
# Gateway 10.45.1g: the .install4j/restarter binary ships and runs, but
# invoking it produces no restarter.log, while the issue #23 reporter's
# 10.45.1j box writes one with install4j action-log content. So the log
# is the fast, unambiguous signal — not the only one. When it is absent
# or stale after a clean exit, the controller spends this long asking
# the agent socket a more direct question: is a live Gateway JVM that we
# did not spawn already answering on it? A yes means Gateway restarted
# itself no matter what install4j logged, and launching a second
# instance is precisely the issue #23 bug.
_AUTO_RESTART_PROBE_SECONDS = int(os.environ.get(
    "AUTO_RESTART_PROBE_SECONDS", "15"))
# After a clean (code-0) exit, keep re-checking restarter.log for this
# long before concluding the exit was not a self-restart: the
# restarter's first write can trail the JVM exit by a couple of seconds
# and monitor_loop polls every 5s.
_AUTO_RESTART_DETECT_GRACE_SECONDS = 5
# After adoption, how long to wait for the API port (session preserved)
# or a login dialog (session not preserved, e.g. IBKR's weekly full
# re-authentication) before falling back to a relaunch.
_AUTO_RESTART_API_TIMEOUT_SECONDS = 180
# Exit status reported by _AdoptedProcess for a JVM the controller did
# not spawn: waitpid() only works on children, so the real code is not
# observable. Deliberately not an int so it can never be mistaken for a
# real exit code.
_EXIT_STATUS_UNOBSERVABLE = "unobservable"


def _compute_adaptive_cooldown(attempt, base_seconds, multiplier, max_seconds):
    """Pure-logic helper: scale the CCP cool-down by restart-attempt index.

    Attempt 1 returns base_seconds; each subsequent attempt multiplies.
    Values of ``attempt <= 0`` are treated as 1 (base duration). Result
    is capped at ``max_seconds`` and returned as ``int`` seconds.

    Split out so the scaling curve is unit-testable without blocking
    on ``time.sleep``.
    """
    safe_attempt = max(int(attempt), 1)
    scaled = int(base_seconds * (multiplier ** (safe_attempt - 1)))
    return min(scaled, max_seconds)


def _apply_ccp_long_cooldown(reason, attempt=1):
    """Sleep long enough for IBKR's CCP rate limiter AND any stranded
    session slots to drain before a JVM restart.

    v0.5.5: adaptive scaling by restart attempt. Each consecutive
    attempt extends the wait by ``CCP_COOLDOWN_MULTIPLIER`` (default
    1.5x), capped at ``CCP_COOLDOWN_MAX_SECONDS`` (default 3600 = 1h).
    Prior versions slept a fixed ``CCP_COOLDOWN_SECONDS`` (default 1200)
    every time, which cleared the rate limiter but could not outlast an
    IBKR-side session-slot hold from a prior unclean teardown — the
    pattern documented in memory/project_ccp_concurrent_session.md
    where lockouts persisted across multiple full escalation cycles.

    Args:
        reason: context string included in the log line.
        attempt: 1-indexed restart-attempt number from the caller's loop
                 (``_escalate_to_jvm_restart``). Defaults to 1 for legacy
                 callers that don't track attempt count.

    Env vars:
        CCP_COOLDOWN_SECONDS     base duration (default 1200)
        CCP_COOLDOWN_MAX_SECONDS cap (default 3600)
        CCP_COOLDOWN_MULTIPLIER  per-attempt multiplier (default 1.5;
                                 set to 1.0 for fixed-duration legacy
                                 behaviour).
    """
    base = int(os.environ.get(
        "CCP_COOLDOWN_SECONDS", str(_CCP_COOLDOWN_SECONDS_DEFAULT)))
    cap = int(os.environ.get(
        "CCP_COOLDOWN_MAX_SECONDS", str(_CCP_COOLDOWN_MAX_SECONDS_DEFAULT)))
    mult = float(os.environ.get(
        "CCP_COOLDOWN_MULTIPLIER", str(_CCP_COOLDOWN_MULTIPLIER_DEFAULT)))
    cool_down_s = _compute_adaptive_cooldown(attempt, base, mult, cap)
    log.warning(
        f"CCP long cool-down ({reason}): sleeping {cool_down_s}s "
        f"(attempt={attempt}, base={base}, mult={mult}, cap={cap}). "
        "Adaptive scaling lets IBKR's rate limiter + any stranded "
        "session slots drain; each restart attempt extends the wait.")
    time.sleep(cool_down_s)


def _is_ibkr_maintenance_window(now=None):
    """v0.5.10: return True if the current wallclock is inside IBKR's
    daily server-side maintenance window.

    Default 23:30-02:00 America/New_York, covering IBKR's published North
    America resets (00:15-01:45 ET Sunday-Friday, 00:00-02:00 ET
    Saturday). The zone is IBKR's, not the container's ``TIME_ZONE``:
    the reset is anchored to the region hosting the account. Override
    both with IBKR_RESET_WINDOWS / IBKR_RESET_TZ.

    ``now`` is an injection seam for tests (pass a tz-aware datetime);
    production callers pass nothing and get the live ET clock.
    """
    if now is None:
        now = datetime.now(ZoneInfo(_CCP_MAINTENANCE_WINDOW_TZ))
    t = now.time()
    for start, end in _RESET_WINDOWS:
        if start <= end:
            if start <= t < end:
                return True
        elif t >= start or t < end:
            return True
    return False


def _apply_maintenance_recovery_delay(reason):
    """v0.5.10: sleep ``CCP_MAINTENANCE_RECOVERY_DELAY_SECONDS`` (default
    480 = 8min) before re-auth so IBKR's auth server has time to drain
    the cooperatively-shutdown session.

    Emits ``ALERT_IBKR_MAINTENANCE_RECOVERY`` (INFO-level grep contract)
    so operators can distinguish this benign delay from a genuine CCP
    cascade. The delay itself is the mitigation — no halt is fired by
    this path; the caller resumes normal recovery after the sleep.
    """
    delay = _CCP_MAINTENANCE_RECOVERY_DELAY_SECONDS
    # Stable grep token for external monitoring. Fires once per recovery
    # path entry. See docs/OBSERVABILITY.md for the contract.
    log.info(
        f"ALERT_IBKR_MAINTENANCE_RECOVERY delay_seconds={delay} "
        f"mode={TRADING_MODE} reason=\"{reason}\"")
    log.warning(
        f"Inside IBKR's reset window ({os.environ.get('IBKR_RESET_WINDOWS', '').strip() or _RESET_WINDOWS_DEFAULT} {_CCP_MAINTENANCE_WINDOW_TZ}); sleeping "
        f"{delay}s before re-auth to let IBKR's auth server drain the "
        f"prior session. Re-auth'ing too quickly during this window "
        f"hits a still-draining server and is silently dropped — the "
        f"2026-04-20/21 CCP-cascade pathology.")
    time.sleep(delay)
    log.info("Maintenance-window delay complete; proceeding with recovery")


# ── Gateway self-restart adoption (issue #23) ──────────────────────────

def _proc_stat(pid):
    """Return ``(state, starttime)`` from ``/proc/<pid>/stat``, or ``None``
    when /proc is unavailable (macOS test hosts) or the entry is gone.

    ``state`` is the kernel's one-letter process state (``Z`` = zombie);
    ``starttime`` is the process start time in clock ticks since boot,
    which together with the PID identifies one process instance.
    """
    try:
        with open(f"/proc/{pid}/stat", "rb") as f:
            raw = f.read().decode("utf-8", errors="replace")
    except (OSError, ValueError):
        return None
    # Field 2 (comm) is parenthesised and may contain spaces; split
    # after the LAST ')' so rest[0] is field 3 (state) and rest[19] is
    # field 22 (starttime).
    try:
        rest = raw[raw.rindex(")") + 2:].split()
        return rest[0], int(rest[19])
    except (ValueError, IndexError):
        return None


class _AdoptedProcess:
    """Popen-shaped stand-in for a Gateway JVM the controller did not spawn.

    Issue #23: after Gateway's own daily auto-restart the replacement JVM
    is install4j's child, not ours, so there is no ``subprocess.Popen``
    to poll. This class gives ``GATEWAY_PROC`` the surface the rest of
    the controller uses (``pid``, ``poll()``, ``returncode``,
    ``terminate()``, ``kill()``, ``wait()``) backed by signals:
    ``os.kill(pid, 0)`` for liveness, plus ``/proc`` (when present) to
    treat a zombie as exited and to detect PID reuse via the kernel
    start time captured at adoption.

    ``waitpid`` only works on children, so the real exit status is not
    observable: ``poll()`` / ``returncode`` report
    ``_EXIT_STATUS_UNOBSERVABLE`` once the process is gone.
    """

    def __init__(self, pid):
        self.pid = int(pid)
        self.returncode = None
        stat = _proc_stat(self.pid)
        self._starttime = stat[1] if stat is not None else None

    def _alive(self):
        # If this PID happens to be our child after all (it isn't in
        # production, where install4j spawned it — but it is in the
        # end-to-end tests), reap it so its exit is observable. On a
        # non-child this raises ECHILD and we fall through to the
        # signal- and /proc-based checks below. Without this, a dead
        # JVM on a host with no /proc looks alive forever and teardown
        # burns its full SIGTERM grace on a process that is already gone.
        try:
            reaped, _ = os.waitpid(self.pid, os.WNOHANG)
            if reaped == self.pid:
                return False
        except (ChildProcessError, OSError):
            pass
        try:
            os.kill(self.pid, 0)
        except ProcessLookupError:
            return False
        except PermissionError:
            # Exists but we may not signal it; let /proc decide below
            # (a PID reused by another user's process fails the
            # start-time check).
            pass
        stat = _proc_stat(self.pid)
        if stat is None:
            return True  # no /proc: trust kill(0)
        state, starttime = stat
        if state in ("Z", "X"):
            return False
        if self._starttime is not None and starttime != self._starttime:
            return False  # PID reused by a different process
        return True

    def poll(self):
        if self.returncode is None and not self._alive():
            self.returncode = _EXIT_STATUS_UNOBSERVABLE
        return self.returncode

    def wait(self, timeout=None):
        deadline = None if timeout is None else time.monotonic() + timeout
        while self.poll() is None:
            if deadline is not None and time.monotonic() >= deadline:
                raise subprocess.TimeoutExpired(f"pid {self.pid}", timeout)
            time.sleep(0.1)
        return self.returncode

    def _signal(self, sig):
        # Identity check first: this PID is not our child, so between
        # adoption and here the kernel could have recycled it onto an
        # unrelated process. _alive() compares /proc start time against
        # the one captured at adoption.
        if not self._alive():
            return
        try:
            os.kill(self.pid, sig)
        except (ProcessLookupError, PermissionError):
            pass

    def terminate(self):
        self._signal(signal.SIGTERM)

    def kill(self):
        self._signal(signal.SIGKILL)

    def __repr__(self):
        return f"<_AdoptedProcess pid={self.pid} returncode={self.returncode!r}>"


def _install4j_restarter_log_path():
    """Path of install4j's ``restarter.log`` for the launcher we spawned
    (``<launcher dir>/.install4j/restarter.log``), or ``None`` if no
    launcher is known."""
    launcher = _GATEWAY_LAUNCHER_PATH or find_gateway_launcher()
    if not launcher:
        return None
    return os.path.join(os.path.dirname(launcher), ".install4j",
                        "restarter.log")


# install4j's restarter is itself a JVM launched from the same
# i4jruntime.jar Gateway uses, and it inherits INSTALL4J_ADD_VM_PARAMS
# from the JVM that spawned it — so it loads OUR agent and binds OUR
# socket for the couple of seconds it lives, answering GET_PID with its
# own PID. Verified 2026-09-06 against the real binary: the restarter
# logged "[gateway-input-agent] listening on <socket>" and reported its
# own pid. Adopting it as if it were Gateway would mean monitoring a
# process that exits immediately.
#
# The discriminator is -Dinstall4j.alternativeLogfile, which the
# restarter sets to point at its own log and Gateway's JVM does not
# carry (confirmed against both live Gateway JVMs in a running
# container). i4jruntime.jar is NOT a discriminator — Gateway's own
# cmdline contains it too.
_INSTALL4J_RESTARTER_MARKER = "install4j.alternativeLogfile"


def _cmdline_is_install4j_restarter(cmdline):
    """True if this process cmdline is install4j's restarter JVM rather
    than Gateway itself. ``cmdline`` is the NUL-joined /proc cmdline."""
    if not cmdline:
        return False
    return _INSTALL4J_RESTARTER_MARKER in cmdline


def _process_cmdline(pid):
    """Return a process's cmdline as a space-joined string, or ``None``
    where /proc is unavailable (macOS) or the process is gone."""
    try:
        with open(f"/proc/{pid}/cmdline", "rb") as f:
            raw = f.read()
    except (OSError, ValueError):
        return None
    return raw.replace(b"\x00", b" ").decode("utf-8", errors="replace")


def _is_install4j_restarter_pid(pid):
    """True only when we can positively identify ``pid`` as install4j's
    restarter. Unknown (no /proc) reads as False so behaviour off Linux
    is unchanged — the adoption retry still covers it."""
    return _cmdline_is_install4j_restarter(_process_cmdline(pid))


def _install4j_restarter_mtime():
    """mtime of install4j's ``restarter.log``, or ``None`` if unreadable."""
    path = _install4j_restarter_log_path()
    if path is None:
        return None
    try:
        return os.stat(path).st_mtime
    except OSError:
        return None


def _install4j_restarter_age(now=None):
    """Seconds since install4j's restarter last wrote its log, or ``None``
    if the log is missing or older than
    ``_AUTO_RESTART_DETECT_WINDOW_SECONDS``.

    install4j having just written that file is the restarter's own
    statement that it ran — the only trigger used for self-restart
    detection (issue #23). A stale file from yesterday's restart, or one
    left behind on a persistent settings volume, is ignored.
    """
    mtime = _install4j_restarter_mtime()
    if mtime is None:
        return None
    age = (time.time() if now is None else now) - mtime
    if age > _AUTO_RESTART_DETECT_WINDOW_SECONDS:
        return None
    if age < -_AUTO_RESTART_DETECT_WINDOW_SECONDS:
        # Dated far in the future — a clock skew between the container
        # and the settings volume, not a restart in progress. Treating
        # it as fresh would send every clean exit through the adoption
        # wait forever.
        return None
    return max(age, 0.0)


def _wait_for_self_restarted_agent(old_pid, timeout, exclude=(),
                                   seen_restarter=None):
    """Poll the agent socket until a JVM with a PID other than ``old_pid``
    (and not in ``exclude``) answers ``GET_PID``. Returns the new PID, or
    ``None`` on timeout.

    ``exclude`` carries PIDs already tried and found dead, so a second
    attempt doesn't re-adopt a transient JVM from the restarter chain.

    install4j's restarter answers on this socket too (it inherits the
    ``-javaagent`` and binds it for the seconds it lives). It is skipped
    rather than adopted, and its PID is appended to ``seen_restarter``
    if a list is passed — seeing it at all is conclusive evidence that
    Gateway is restarting itself.

    The self-restarted JVM inherits ``INSTALL4J_ADD_VM_PARAMS`` through
    the restarter, so it loads the same ``-javaagent`` with the same
    socket path, and the agent's own ``Files.deleteIfExists`` + bind
    replaces the dead JVM's socket file. This side must NOT unlink the
    socket: doing so after the new agent has bound would orphan its
    listener and the controller could never reach the adopted JVM.
    """
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if agent_wait_ready(timeout=1):
            pid = agent_get_pid()
            if (pid is not None and pid != old_pid and pid not in exclude
                    and pid > 1 and pid != os.getpid()):
                if _is_install4j_restarter_pid(pid):
                    if seen_restarter is not None and pid not in seen_restarter:
                        seen_restarter.append(pid)
                        log.info(f"AUTORESTART: install4j's restarter "
                                 f"(pid={pid}) is holding the agent socket — "
                                 "Gateway is definitely restarting itself; "
                                 "waiting for the Gateway JVM it launches")
                else:
                    return pid
        time.sleep(0.5)
    return None


def _adopt_self_restarted_gateway(reason, *, exit_code=None):
    """Issue #23: if the JVM exit that triggered recovery was Gateway's
    own daily auto-restart, adopt the instance install4j is bringing up
    instead of launching a second one.

    Returns True when the adopted Gateway is serving again (API port
    open because the session was preserved, or re-login driven when it
    was not) and monitoring can resume. Returns False to fall through
    to the controller-driven restart. By then ``GATEWAY_PROC`` may point
    at the adopted instance so the fallback's teardown terminates it
    first, instead of no-op'ing on the dead ``Popen`` and leaving two
    Gateways up. Expected failure modes never raise.

    Emits ``ALERT_AUTO_RESTART`` with ``status=adopted`` (INFO) or one
    of the ``failed_*`` statuses (WARNING); see docs/OBSERVABILITY.md.
    """
    global GATEWAY_PROC, CURRENT_APP, JVM_PID, _command_server_app
    global _LAST_ADOPTED_RESTART_MTIME

    old_pid = JVM_PID
    clean_exit = exit_code in (0, _EXIT_STATUS_UNOBSERVABLE)
    age = _install4j_restarter_age()
    if age is None and clean_exit:
        grace_deadline = time.monotonic() + _AUTO_RESTART_DETECT_GRACE_SECONDS
        while age is None and time.monotonic() < grace_deadline:
            time.sleep(0.5)
            age = _install4j_restarter_age()

    first_candidate = None
    seen_restarter = []
    if age is None:
        # No usable restarter.log. On a crash, stop here — that is a
        # genuine failure and the relaunch path owns it.
        if not clean_exit or _AUTO_RESTART_PROBE_SECONDS <= 0:
            return False
        # On a clean exit, ask the agent socket directly before
        # relaunching. Only a running JVM answers it (a dead one leaves
        # the socket file behind but nothing listening), so a reply from
        # a PID that is not ours is positive evidence of a self-restart.
        log.info("AUTORESTART: no fresh install4j restarter.log; probing "
                 f"{AGENT_SOCKET} for up to {_AUTO_RESTART_PROBE_SECONDS}s "
                 "in case Gateway restarted itself anyway")
        first_candidate = _wait_for_self_restarted_agent(
            old_pid, _AUTO_RESTART_PROBE_SECONDS,
            seen_restarter=seen_restarter)
        if first_candidate is None and not seen_restarter:
            return False
        detected_via = ("install4j_restarter" if seen_restarter
                        else "agent_socket")
        who = ("install4j's restarter" if seen_restarter
               else "a Gateway JVM we did not spawn")
        log.warning(f"AUTORESTART: {who} is on {AGENT_SOCKET} — treating "
                    "this as Gateway's own restart and NOT launching a "
                    "second instance")
    else:
        detected_via = "restarter_log"

    mtime = _install4j_restarter_mtime()
    if (detected_via == "restarter_log" and mtime is not None
            and mtime == _LAST_ADOPTED_RESTART_MTIME):
        # Same restart we already handled: the adopted JVM came up and
        # then died inside the 120s freshness window. That's a failure
        # of the restarted instance, not a restart in progress — go
        # straight to the controller-driven relaunch.
        log.info("AUTORESTART: restarter.log unchanged since the last "
                 "adoption attempt — treating this exit as a real "
                 "failure, not another self-restart")
        return False

    if detected_via == "restarter_log":
        log.info(f"AUTORESTART: install4j restarter ran {age:.0f}s ago "
                 f"({_install4j_restarter_log_path()})")
        log.warning(f"AUTORESTART: Gateway is restarting itself — NOT "
                    f"launching a second instance (old pid={old_pid}); "
                    f"waiting up to {_AUTO_RESTART_ADOPT_TIMEOUT_SECONDS}s "
                    f"for the new JVM's agent on {AGENT_SOCKET}")
        _LAST_ADOPTED_RESTART_MTIME = mtime
    t0 = time.monotonic()

    def _alert(level, status, new_pid, why):
        level(f"ALERT_AUTO_RESTART mode={TRADING_MODE} status={status} "
              f"detected_via={detected_via} "
              f"old_pid={old_pid} new_pid={new_pid} "
              f"elapsed_seconds={time.monotonic() - t0:.0f} "
              f"reason=\"{why}\"")

    api_port = api_port_for_mode()
    # install4j's restarter chain can briefly show a JVM that isn't the
    # replacement Gateway. If the first candidate dies before its API
    # port opens, try once more for whatever is left of the budget
    # rather than falling straight through to a relaunch.
    tried = set()
    adopt_deadline = time.monotonic() + _AUTO_RESTART_ADOPT_TIMEOUT_SECONDS
    while True:
        if first_candidate is not None:
            new_pid, first_candidate = first_candidate, None
        else:
            wait_budget = max(1.0, adopt_deadline - time.monotonic())
            new_pid = _wait_for_self_restarted_agent(
                old_pid, wait_budget, exclude=tried,
                seen_restarter=seen_restarter)
        if new_pid is None or new_pid in tried:
            _alert(log.warning, "failed_no_agent", "none",
                   f"no new Gateway JVM answered on {AGENT_SOCKET} within "
                   f"{_AUTO_RESTART_ADOPT_TIMEOUT_SECONDS}s; falling back to a "
                   "controller-driven relaunch")
            return False
        tried.add(new_pid)

        # Repoint GATEWAY_PROC BEFORE the API-port wait: if the port never
        # opens, the fallback's _teardown_jvm_for_restart then cleanly kills
        # the adopted instance before relaunching.
        GATEWAY_PROC = _AdoptedProcess(new_pid)
        global _JVM_STARTED_WALL, _existing_session_continued_this_jvm
        global _existing_session_closed_after_continue
        _JVM_STARTED_WALL = datetime.now()
        _existing_session_continued_this_jvm = False
        _existing_session_closed_after_continue = False
        JVM_PID = new_pid
        new_app = find_app(APP_NAME_CANDIDATES, timeout=120, match_pid=new_pid)
        CURRENT_APP = new_app
        _command_server_app = new_app
        log.info(f"AUTORESTART: adopted Gateway pid={new_pid} (was {old_pid}) "
                 f"after {time.monotonic() - t0:.0f}s")

        deadline = time.monotonic() + _AUTO_RESTART_API_TIMEOUT_SECONDS
        last_status = time.monotonic()
        died = False
        own_login_seen = False
        ready_polls = 0
        while time.monotonic() < deadline:
            if GATEWAY_PROC.poll() is not None:
                died = True
                break
            if is_api_port_open(api_port):
                _alert(log.info, "adopted", new_pid,
                       f"API port {api_port} open — session preserved across "
                       "Gateway's auto-restart, no login and no second factor "
                       "required")
                signal_ready()
                return True
            texts, buttons = agent_list()
            login_window = "Username" in texts or "Password" in texts
            state = agent_login_state() if login_window else None
            if state == "busy":
                # Gateway restores the session it saved before restarting
                # and logs in with it on its own, with its fields greyed out
                # until that login finishes. Typing over it fails (#76),
                # so wait: the API port opens if the session carried over.
                if not own_login_seen:
                    log.info("AUTORESTART: Gateway is logging in with the "
                             "session it saved before restarting (its login "
                             "fields are greyed out); waiting for it")
                    own_login_seen = True
                ready_polls = 0
            elif state == "ready" and ready_polls == 0:
                # Gateway may start that login a moment after showing the
                # window, so look once more before treating it as waiting
                # for us.
                ready_polls = 1
            elif login_window:
                if own_login_seen:
                    log.info("AUTORESTART: Gateway's own login with its saved "
                             "session didn't complete and the login fields "
                             "are enabled again; re-driving login")
                else:
                    log.info("AUTORESTART: login dialog on the self-restarted "
                             "Gateway — the session was not preserved (e.g. "
                             "IBKR's weekly full re-authentication); "
                             "re-driving login")
                # AUTO_RESTART_TIME commonly sits inside IBKR's nightly
                # maintenance window. Adoption itself needs no auth, but
                # this branch does, so the v0.5.10 drain guard applies.
                if _is_ibkr_maintenance_window():
                    _apply_maintenance_recovery_delay(
                        "self-restarted Gateway needs a full re-login")
                if _redrive_login(new_app):
                    _alert(log.info, "adopted", new_pid,
                           "session not preserved; full login re-driven on the "
                           "adopted JVM")
                    return True
                _alert(log.warning, "failed_login", new_pid,
                       "login on the adopted Gateway failed; falling back to a "
                       "controller-driven relaunch")
                return False
            else:
                ready_polls = 0
            # Same opportunistic disclaimer handling as wait_for_api_port.
            for btn in SAFE_DISMISS_BUTTONS:
                if btn in buttons:
                    log.info(f"  dismissing disclaimer {btn!r}")
                    agent_click(btn)
                    time.sleep(0.5)
            now = time.monotonic()
            if now - last_status > 10:
                log.info(_redact_logs(
                    f"AUTORESTART: API port {api_port} still closed at "
                    f"t+{now - t0:.0f}s; windows={agent_windows()}"))
                last_status = now
            time.sleep(1)

        if died and time.monotonic() < adopt_deadline:
            log.warning(f"AUTORESTART: adopted pid={new_pid} exited before its "
                        "API port opened; looking for another instance from "
                        "the restarter chain")
            continue
        if died:
            _alert(log.warning, "failed_jvm_exited", new_pid,
                   "adopted Gateway JVM exited before its API port opened; "
                   "falling back to a controller-driven relaunch")
            return False
        if own_login_seen:
            why = (f"API port {api_port} did not open within "
                   f"{_AUTO_RESTART_API_TIMEOUT_SECONDS}s of adoption while "
                   "Gateway was logging in with its saved session; falling "
                   "back to a controller-driven relaunch")
        else:
            why = (f"API port {api_port} did not open within "
                   f"{_AUTO_RESTART_API_TIMEOUT_SECONDS}s of adoption and no "
                   "login dialog appeared; falling back to a "
                   "controller-driven relaunch")
        _alert(log.warning, "failed_api_timeout", new_pid, why)
        return False


def _recover_jvm_or_escalate(reason, *, exit_code=None):
    """v0.4.7: Attempt a fast in-place JVM restart; on failure fall
    through to ``_escalate_to_jvm_restart`` (long CCP cool-down).

    Used by ``monitor_loop`` on paths where the old code called
    ``sys.exit(rc)``. In dual-mode those exits were silent no-ops —
    the container stayed up on the *other* mode's PID and this mode's
    JVM stayed dead forever (same trap v0.4.5/v0.4.6 fixed for the
    CCP-lockout paths). 2026-04-17 validation: live JVM exited
    cleanly (code 0) 18min after container start, ``monitor_loop``
    ``sys.exit``'d, live port 4001 was refused from outside the
    container while the container itself kept serving paper.

    Fast path first because an unexpected clean JVM exit (IBKR session
    kick, auto-logoff) usually isn't CCP-related — ``do_restart_in_place``
    will relaunch and log in immediately, no 20min wait. If it does
    fail (CCP lockout on the relaunched JVM, for instance), we fall
    through to the silent-cool-down escalation.

    Never returns False in production: it returns True on recovery, and
    if everything fails ``_escalate_to_jvm_restart`` halts (since v0.13.0),
    which never returns.

    v0.5.10: maintenance-window guard. When ``exit_code == 0`` AND
    wallclock is inside IBKR's reset window (23:30-02:00 ET by default),
    sleep ``CCP_MAINTENANCE_RECOVERY_DELAY_SECONDS`` (default 480)
    before attempting the fast restart. The 2026-04-20/21 incident
    showed that re-auth ~8s after a code-0 exit in this window hits
    IBKR's still-draining auth server and is silently dropped, setting
    off a CCP-lockout cascade on both modes. Non-zero exits bypass the
    guard — they're crashes, not maintenance shutdowns, and should
    recover fast.

    Issue #23: before either of those, if the old JVM is gone and
    install4j's restarter just ran, ``_adopt_self_restarted_gateway``
    adopts the instance Gateway is bringing up itself (no relaunch, no
    login, no second factor). Ordered ahead of the maintenance guard on
    purpose: AUTO_RESTART_TIME commonly sits inside that window, and an
    8-minute sleep would be pointless when there is nothing to re-auth.
    ``_EXIT_STATUS_UNOBSERVABLE`` (an adopted JVM's exit — its real code
    can't be observed) is treated like 0 by the guard: the conservative
    choice, since the common cause of an adopted JVM exiting is the same
    cooperative shutdown.
    """
    if _AUTO_RESTART_ADOPT and (GATEWAY_PROC is None
                                or GATEWAY_PROC.poll() is not None):
        try:
            if _adopt_self_restarted_gateway(reason, exit_code=exit_code):
                log.info("Recovery: adopted Gateway's self-restarted JVM; "
                         "no relaunch needed")
                return True
        except Exception as e:
            log.error(f"Recovery: self-restart adoption raised "
                      f"{type(e).__name__}: {e}; falling back to relaunch")
        if isinstance(GATEWAY_PROC, _AdoptedProcess) and GATEWAY_PROC.poll() is None:
            # Half-adopted instance is still up: take it down now so no
            # second Gateway lingers through any delay below, and the
            # relaunch starts from a clean slate.
            log.warning("AUTORESTART: tearing down the adopted Gateway "
                        f"pid={GATEWAY_PROC.pid} before the fallback relaunch")
            _teardown_jvm_for_restart()

    if exit_code in (0, _EXIT_STATUS_UNOBSERVABLE) and _is_ibkr_maintenance_window():
        _apply_maintenance_recovery_delay(reason)

    log.warning(f"Recovery: {reason}. Trying fast in-place restart first.")
    try:
        if do_restart_in_place():
            log.info("Recovery: fast restart succeeded")
            return True
    except Exception as e:
        log.error(f"Recovery: do_restart_in_place raised "
                  f"{type(e).__name__}: {e}")
    log.warning("Recovery: fast restart failed; escalating to "
                "long-cool-down JVM restart")
    return _escalate_to_jvm_restart(reason)


def _quiet_jvm_for_halt():
    """Release the IBKR session slot and stop the JVM before halting.

    Clean UI logout first, then SIGTERM with 30 s grace so Gateway's
    shutdown hook can close the IBKR session; SIGKILL strands the slot and
    extends IBKR's server-side zombie timer (2026-04-19), so it is only the
    last resort for a JVM that ignores SIGTERM. That is the same end the
    JVM met when the controller exited and the container stopped, and the
    halt promises no further login attempts, which only a dead JVM keeps.
    Lifted from the default CCP halt (2026-04-27) so the opt-in path, which
    used to exit without releasing the slot, uses it too.
    """
    if GATEWAY_PROC is None or GATEWAY_PROC.poll() is not None:
        return
    pid = GATEWAY_PROC.pid
    clean_success, clean_status, clean_reason = _attempt_clean_logout()
    log.info(
        f"ALERT_CLEAN_LOGOUT mode={TRADING_MODE} pid={pid} "
        f"status={clean_status} reason=\"{clean_reason}\"")
    if clean_success:
        return
    log.warning("Clean logout failed; falling back to JVM SIGTERM with 30s "
                "grace before halting")
    try:
        GATEWAY_PROC.terminate()
        GATEWAY_PROC.wait(timeout=30)
    except subprocess.TimeoutExpired:
        log.error("JVM did not exit within 30s of SIGTERM; killing it so "
                  "nothing keeps talking to IBKR while halted")
        GATEWAY_PROC.kill()


def _escalate_to_jvm_restart(reason):
    """v0.4.5: Dual-mode-aware escape hatch for CCP lockout.

    Replaces the old ``sys.exit(1)`` terminus on paths where in-JVM
    relogin is exhausted (max attempts) or impossible (disposed login
    frame, v0.4.4). In dual-mode the container won't restart on
    sys.exit because ``run.sh`` waits on both live+paper controller
    PIDs and only the exiting mode's process dies — the container
    stays up on the other mode's PID and this mode's JVM stays dead
    forever. That leaves any external API client connecting to a
    dangling socat that ECONNREFUSEs every request.

    v0.4.6 sequencing (kill first, THEN cool-down, THEN relaunch):
    kill this mode's Gateway JVM, sleep the long CCP cool-down
    (default 20min) with NO JVM running on these credentials, then
    launch a fresh JVM and re-drive login. If the JVM is kept alive
    during the cool-down its internal "Attempt N: connecting to
    server" retry loop keeps hitting IBKR's auth server and keeps
    the CCP rate limiter armed — observed 2026-04-16 with v0.4.5,
    where a 20min cool-down with the JVM alive failed to clear CCP
    on the relaunch. v0.4.6 makes the cool-down genuinely silent.

    Not the v0.4.0 bug: v0.4.0 was kill+relaunch+retry with 60-600s
    gaps, too short for CCP to clear and each fresh handshake rearmed
    the limiter. v0.4.6 keeps the 20min duration but positions it so
    the JVM is dead for the full wait.

    Returns True on successful JVM restart. When restarting keeps
    failing, or is disabled (the default), it HALTS — stays up with state
    HALTED — rather than exiting (field report 2026-10-01: an exit restarts
    the container, which re-authenticates against a slot still held).

    v0.5.9: halt-by-default via ``CCP_LOCKOUT_MAX_JVM_RESTARTS``. When
    that env var is 0 (the new default), this function emits
    ``ALERT_CCP_PERSISTENT_HALT`` straight away, with no teardown and
    relaunch loop (since v0.13.0 it then releases the session and halts
    rather than exiting). This prevents the 2026-04-19 re-stranding pattern — each
    prior escalation cycle's SIGKILL teardown was extending IBKR's
    server-side zombie-slot timer, compounding the lockout we were
    trying to clear. When set to a positive integer, restores the
    pre-v0.5.9 loop capped at that many attempts (supersedes
    ``_JVM_RESTART_MAX_ATTEMPTS``).
    """
    if _CCP_LOCKOUT_MAX_JVM_RESTARTS <= 0:
        log.error(
            "CCP-lockout-triggered JVM restart is disabled (default since "
            "v0.5.9). Each SIGKILL teardown re-strands the IBKR session "
            "slot and extends the server-side zombie timer; the safest "
            "recovery is to halt and let an operator investigate. Set "
            "CCP_LOCKOUT_MAX_JVM_RESTARTS to a positive integer to "
            "restore the pre-v0.5.9 auto-restart loop.")
        # Release the slot cleanly first (2026-04-27): a JVM killed by
        # process-tree teardown strands it for hours.
        _quiet_jvm_for_halt()
        # Stable grep token for external monitoring. Emitted exactly once
        # per terminal halt. See docs/OBSERVABILITY.md for the grep-contract.
        log.error(
            f"ALERT_CCP_PERSISTENT_HALT mode={TRADING_MODE} "
            f"reason=\"{reason}\" "
            f"remediation=\"log into IBKR Mobile as this username to "
            f"force-log-out the held TWS/Gateway slot (IBKR Client "
            f"Portal login does NOT kick the slot — confirmed in "
            f"production), then restart the container\"")
        # HALT, don't exit. Exiting hands the container to its restart
        # policy, and since v0.12.0 a live exit restarts the whole
        # container: a fresh container re-authenticates at once against a
        # slot that is still held, which is the 2026-04-19 loop this halt
        # exists to prevent (field report 2026-10-01). Staying up keeps
        # the auth pipe closed until a person clears the slot.
        _halt_for_operator(
            "persistent CCP lockout; see ALERT_CCP_PERSISTENT_HALT above. "
            "Clear the held slot, then restart the container")
        return False

    cap = _CCP_LOCKOUT_MAX_JVM_RESTARTS
    with _restart_lock:
        return _escalate_restart_loop(reason, cap)


def _escalate_restart_loop(reason, cap):
    """The opt-in CCP restart loop, under the relaunch lock."""
    for attempt in range(1, cap + 1):
        log.warning(f"JVM restart attempt {attempt}/{cap}: "
                    "tearing down JVM before long cool-down (v0.4.6 silent cool-down)")
        _teardown_jvm_for_restart()
        _apply_ccp_long_cooldown(
            f"{reason}; JVM restart {attempt}/{cap}",
            attempt=attempt,
        )
        log.warning(f"JVM restart attempt {attempt}/{cap}: "
                    "cool-down complete, launching fresh JVM")
        if _relaunch_and_login_in_place():
            log.info("JVM restart succeeded; API port is open")
            _reset_ccp_backoff()
            return True
        log.error(f"JVM restart attempt {attempt} failed")
    log.error(f"JVM restart limit ({cap}) exhausted "
              "after silent cool-downs; halting")
    # Stop the last relaunched JVM retrying against IBKR, and release its
    # slot, before going quiet.
    _quiet_jvm_for_halt()
    # Stable grep token for external monitoring. Emitted exactly once per
    # terminal escalation. See docs/OBSERVABILITY.md for the grep-contract.
    log.error(
        f"ALERT_JVM_RESTART_EXHAUSTED mode={TRADING_MODE} "
        f"attempts={cap} "
        f"reason=\"{reason}\"")
    # Halt rather than exit, for the same reason as the default path.
    _halt_for_operator(
        f"{cap} JVM restarts did not clear the CCP lockout; see "
        "ALERT_JVM_RESTART_EXHAUSTED above")
    return False


def _looks_like_disposed_shell(windows):
    """True when the visible window set matches Gateway's post-CCP
    disposed-login-frame state.

    Observed shape (v0.4.3 live validation from production logs):
    a single non-modal top-level Window titled ``IBKR Gateway`` with no
    JPasswordField anywhere in its tree. Gateway's main application
    shell has rendered with its File/Configure/Help menu bar and the
    "API Server: disconnected" status labels — the login frame has
    been disposed, not occluded.

    In this state ``LoginManager.initiateLogin(capturedLoginFrame)`` is
    a silent no-op because the captured reference points at a disposed
    Window. In-JVM relogin cannot recover; the only path forward is a
    full JVM restart via container-level kill+relaunch.

    The caller should only trust this signature AFTER a short
    ``agent_wait_login_frame`` probe has already failed — that probe
    confirms there is no JPasswordField-bearing Window currently up,
    which distinguishes this case from the stuck-connecting retry
    (where the login frame IS present but a modal progress dialog is
    on top).
    """
    if len(windows) != 1:
        return False
    _wtype, title, modal = windows[0]
    return not modal and "IBKR Gateway" in title


def attempt_inplace_relogin(app):
    """In-JVM relogin primitive — match IBC's
    ``getLoginHandler().initiateLogin(getLoginFrame())`` semantics from
    outside the JVM via the existing AT-SPI path.

    Does NOT terminate GATEWAY_PROC, does NOT unlink AGENT_SOCKET, does
    NOT call launch_gateway. CURRENT_APP / GATEWAY_PROC / JVM_PID stay
    intact across the call.

    Why: killing + relaunching the Gateway JVM creates a fresh TCP/TLS
    session that IBKR's auth server treats as a new handshake, which
    keeps the CCP rate limiter armed and the lockout cycle open. IBC
    has worked in production for years by staying in one JVM and just
    re-invoking ``initiateLogin`` on the existing login frame. See the
    v0.4.0 CHANGELOG entry for the full rationale.

    Strategy (AT-SPI side):
      1. Scan open modal windows. Dismiss known login-failure error
         modals via OK/Close. Never click on a 'Connecting to server'
         progress dialog — clicking OK cancels the login.
      2. Wait up to 30s for the login form to redisplay by polling for
         the password text field (same signal handle_login uses).
      3. Re-drive handle_login(app) on the same app reference. Return
         its result.

    Returns True if handle_login succeeded, False if the login frame
    never reappeared or handle_login failed. Callers typically couple
    this with _apply_ccp_backoff() BEFORE the call for spacing.
    """
    log.warning("In-JVM relogin attempt (no JVM restart — matches "
                "IBC's LoginManager.initiateLogin semantics)")

    # 1. Dismiss error modals only. Progress dialogs are off-limits.
    try:
        windows = agent_windows()
    except Exception as e:
        log.warning(f"In-JVM relogin: agent_windows failed ({type(e).__name__}: {e}); "
                    "skipping modal dismissal and going straight to login-frame wait")
        windows = []
    for _wtype, title, modal in windows:
        if not modal:
            continue
        tl = title.lower()
        # Leave any in-flight progress dialog alone — OK cancels login.
        if "connecting to server" in tl:
            log.info(f"In-JVM relogin: leaving progress dialog intact: {title!r}")
            continue
        # Inspect body text to decide whether this is a dismissible error.
        try:
            body = agent_window(title)
        except Exception:
            body = ""
        body_lower = body.lower()
        # Split error markers into credential-rejection vs connection
        # failures. Credential rejection warrants the stable
        # ALERT_LOGIN_FAILED grep-contract token so monitors can tell a
        # wrong-password account lockout apart from an IBKR silent
        # cooldown (ALERT_CCP_PERSISTENT). Connection failures stay
        # un-alerted here because they're covered by the CCP backoff
        # path upstream.
        credential_error_markers = (
            "login failed",
            "login error",
            "authentication failed",
        )
        network_error_markers = (
            "could not be performed",
            "unable to connect",
            "server cannot be reached",
        )
        # Halt only on Gateway's explicit wording: the generic markers can
        # also come from a second-factor failure.
        is_rejected = _detect_bad_credentials(body)
        is_credential_error = (
            any(m in body_lower for m in credential_error_markers)
            or is_rejected)
        is_network_error = any(
            m in body_lower for m in network_error_markers)
        if is_credential_error or is_network_error:
            if is_credential_error:
                log.error(
                    f"ALERT_LOGIN_FAILED mode={TRADING_MODE} "
                    f"reason=\"bad-credentials\" "
                    f"suggested_action=\"Gateway surfaced a credential-"
                    f"rejection modal; verify TWS_USERID / TWS_PASSWORD "
                    f"(or _PAPER variants) and update env if password "
                    f"was rotated in IBKR Account Settings\"")
            log.info(f"In-JVM relogin: dismissing error modal: {title!r}")
            for btn in ("OK", "Close"):
                if agent_click_in_window(title, btn):
                    break
            if is_rejected:
                _halt_credentials_rejected()

    # 2. Wait for the login frame to redisplay, but short-circuit the
    # wait when we can tell in-JVM relogin is impossible.
    #
    # v0.4.3: use the Java agent's Swing-type-based lookup
    # (findLoginFrame → showing Window containing JPasswordField, plus
    # no-modal-blocking check) instead of pyatspi
    # ``wait_for(app, "password text")``. AT-SPI filters the login
    # frame's role while Gateway's "Attempt N: connecting to server"
    # modal is up, causing the old 30s wait to time out before
    # Gateway's internal retry self-cleared (typical ~60s). 120s
    # covers one full retry cycle with margin.
    #
    # v0.4.4: probe first with a short 2s timeout. If that fails AND
    # the visible windows match the disposed-shell signature (single
    # non-modal "IBKR Gateway" frame, no JPasswordField), bail
    # immediately and return False. `wait_for_api_port_with_retry`
    # treats that as an in-JVM-relogin failure and escalates to
    # container-level kill+relaunch, which is the only recovery path
    # once Gateway has disposed the login frame. Without this
    # short-circuit we burn 120s per attempt × 8 attempts = 16min
    # waiting for a frame that will never come back.
    log.info("In-JVM relogin: probing for login frame (2s) before full wait")
    if agent_wait_login_frame(timeout_ms=2_000):
        pass  # login frame already interactable, skip to step 3
    else:
        try:
            probe_windows = agent_windows()
        except Exception as e:
            log.warning(f"In-JVM relogin: agent_windows probe failed "
                        f"({type(e).__name__}: {e}); proceeding with full 120s wait")
            probe_windows = None
        if probe_windows is not None and _looks_like_disposed_shell(probe_windows):
            log.error("In-JVM relogin unavailable: login frame disposed "
                      "(post-CCP disconnected shell detected).")
            log.error(f"  windows: {probe_windows}")
            log.error("  Escalating to container-level recovery "
                      "(kill+relaunch JVM — the captured login frame "
                      "reference is a disposed Window, initiateLogin() "
                      "on it is a no-op).")
            return False
        log.info("In-JVM relogin: short probe failed but no disposed-shell "
                 "signature; waiting up to 120s for login frame")
        if not agent_wait_login_frame(timeout_ms=120_000):
            log.error("In-JVM relogin: login frame never became interactable within 120s")
            try:
                windows = agent_windows()
                log.error(f"  windows at timeout: {windows}")
            except Exception as e:
                log.error(f"  windows dump failed: {type(e).__name__}: {e}")
            return False

    # 3. Same app, same JVM — just re-drive handle_login.
    log.info("In-JVM relogin: login frame is up, re-driving handle_login "
             "on the existing JVM")
    return handle_login(app)


def wait_for_api_port_with_retry(app, port_timeout=180,
                                 max_attempts=_INPLACE_RELOGIN_MAX_ATTEMPTS):
    """wait_for_api_port wrapped in an in-JVM relogin retry loop.

    v0.4.1: the v0.4.0 CCP-lockout loop in main() only catches the
    launcher.log ``AuthTimeoutMonitor-CCP: Timeout!`` signature.
    Stuck-connecting mode — Gateway's login dialog stuck in its
    internal "connecting to server (trying for another N seconds)"
    retry — emits no Timeout! line, so it slipped past the v0.4.0
    outer poll at main():2383 and flowed into handle_2fa, which did
    a single RELOGIN_AFTER_TWOFA_TIMEOUT re-drive and then fell
    through to wait_for_api_port with no further retry. On timeout,
    the controller exited and the JVM was orphaned.

    This helper catches both failure modes at the final indicator
    (the API port). If the port doesn't open and either CCP-Timeout
    or stuck-connecting is currently visible, back off and relogin
    in-JVM on the same app reference. Capped at ``max_attempts``
    relogins; past that the controller exits for container-level
    recovery.

    Does NOT call launch_gateway, terminate GATEWAY_PROC, or unlink
    AGENT_SOCKET. The retry primitive is ``attempt_inplace_relogin``.

    Returns True when the API port opens. Calls ``sys.exit(1)`` on
    terminal failure (no lockout signature) or cap exhaustion —
    matching the behavior of the existing v0.4.0 CCP-lockout loop
    in main().
    """
    if wait_for_api_port(timeout=port_timeout):
        _reset_ccp_backoff()
        return True
    for attempt in range(1, max_attempts + 1):
        ccp = _detect_ccp_lockout(timeout=25)
        stuck = _detect_login_stuck_connecting()
        if not (ccp or stuck):
            # No lockout signature — this is a terminal failure
            # (wrong credentials, wrong server, network problem).
            # Preserve the v0.4.0 diagnostic dump for operators.
            log.error("API port never opened and no lockout signature "
                      "detected — treating as terminal failure (wrong "
                      "creds, wrong server, or network)")
            _diagnose_login_failure()
            log.error("Final state dump:")
            log.error(_redact_logs(f"  windows: {agent_windows()}"))
            labels = agent_labels()
            for wtitle, text in labels[:30]:
                log.error(_redact_logs(f"  label [{wtitle}] {text!r}"))
            if _recover_mode_in_place("the API port never opened after login"):
                return True
            sys.exit(1)
        log.warning(f"API port timeout with lockout signature "
                    f"(ccp_timeout={ccp}, stuck_connecting={stuck}); "
                    f"in-JVM relogin attempt {attempt}/{max_attempts}")
        _apply_ccp_backoff()
        if not attempt_inplace_relogin(app):
            log.error("In-JVM relogin failed during API-port retry "
                      "(login frame disposed or handle_login failed); "
                      "escalating to long-cool-down JVM restart (v0.4.5)")
            return _escalate_to_jvm_restart("in-JVM relogin returned False")
        if wait_for_api_port(timeout=port_timeout):
            _reset_ccp_backoff()
            return True
    log.error(f"CCP lockout persists after {max_attempts} in-JVM "
              "relogin attempts at API-port wait; escalating to "
              "long-cool-down JVM restart (v0.4.5 — dual-mode container's "
              "run.sh does NOT restart on sys.exit, so we self-heal)")
    return _escalate_to_jvm_restart(f"{max_attempts} in-JVM relogin attempts exhausted")


def _sibling_mode_ready():
    """In dual mode, True when the other trading mode has signalled
    readiness in this container.

    That proves the network path to IBKR works. It does not prove this
    mode's credentials — live and paper log in with different accounts —
    so the diagnosis says only what it shows.
    """
    for mine, other in (("live", "paper"), ("paper", "live")):
        if READY_FILE.endswith(f"_{mine}"):
            return os.path.exists(READY_FILE[: -len(mine)] + other)
    return False


def _diagnose_login_failure():
    """Parse Gateway's launcher.log to turn a generic 'API port never
    opened' failure into a specific, actionable error message.

    Gateway's auth protocol on the wire:
      1. TCP connect + SSL handshake
      2. Gateway logs 'Authenticating'
      3. Server replies with 'Received NS_AUTH_START'
      4. Gateway sends credentials
      5. Server replies with 'PostAuthenticate' or an error dialog
      6. Gateway opens the API port

    Three observed failure modes look identical at step 6 but leave
    different fingerprints in launcher.log:

      a) IBKR silent cooldown / wrong server / network block:
         'Authenticating' appears, NS_AUTH_START never appears,
         'AuthTimeoutMonitor-CCP: Timeout!' fires 20 seconds later.
         Nothing from the server — it's ignoring us or we can't
         reach it.

      b) Handshake answered, login never completed:
         'Authenticating' appears, 'NS_AUTH_START' appears, but
         PostAuthenticate never happens. With a CCP Timeout! after it,
         that is the credentials being rejected. WITHOUT one, it is
         "post-auth-no-progress", which is not reliably a credential
         problem: a 2026-09-30 field report saw the same account fail
         this way twice and succeed on the third attempt with no
         change at all. Treat it as retryable first.

      c) Never reached Authenticating:
         Usually means the SSL handshake failed, the server was
         unreachable, or the install4j launcher refused to start.
         Look at the tail of launcher.log for the specific error.

    We print a targeted error message for whichever case fits.
    """
    launcher_log = os.path.join(JTS_CONFIG_DIR, "launcher.log")
    if not os.path.isfile(launcher_log):
        log.error(f"Diagnosis: launcher.log not found at {launcher_log} — "
                  "Gateway may have failed before creating its log directory. "
                  "Check Xvfb, the install4j launcher path, and the ATK bridge "
                  "setup in $JAVA_HOME.")
        return

    try:
        with open(launcher_log, encoding="utf-8", errors="replace") as f:
            content = f.read()
    except Exception as e:
        log.error(f"Diagnosis: couldn't read {launcher_log}: {type(e).__name__}: {e}")
        return

    has_authenticating = "Authenticating" in content
    # Judge the latest attempt only: the text from the last
    # "Authenticating" on. An earlier attempt's lines, or SSL errors from
    # other requests before it, mustn't decide (issue #63). A failure that
    # never reached "Authenticating" still lands after the previous one.
    latest = content[content.rfind("Authenticating"):] if has_authenticating else content
    has_ns_auth_start = "NS_AUTH_START" in latest
    has_timeout = "AuthTimeoutMonitor-CCP: Timeout!" in latest
    has_ssl_fail = "SSLHandshakeException" in latest or "Remote host terminated" in latest
    has_compete = ("Competition detected" in latest
                   or "competing session" in latest)

    if has_compete:
        # Issue #63: this used to fall through to the SSL diagnosis below,
        # because SSL errors from other requests are common in the log.
        log.error("Diagnosis: IBKR reported a COMPETING SESSION: the same "
                  "username is logged in elsewhere (another Gateway or TWS, "
                  "Client Portal on the web, or IBKR Mobile).")
        log.error("  EXISTING_SESSION_DETECTED_ACTION=primary takes the "
                  "session over; manual leaves Gateway's dialog for you. Or "
                  "log out of the other session.")
        return

    if has_ssl_fail:
        log.error("Diagnosis: SSL HANDSHAKE FAILED. This usually means "
                  "TWS_SERVER / TWS_SERVER_PAPER points at the wrong "
                  "regional server for this account, OR Gateway's "
                  "SupportsSSL cache is invalid.")
        log.error("  Check controller/docs/BOOTSTRAP.md for how to find "
                  "your account's correct regional server.")
        return

    if not has_authenticating:
        log.error("Diagnosis: Gateway NEVER reached the Authenticating state.")
        log.error("  Possible causes: SSL handshake failed before auth, "
                  "install4j launcher didn't start properly, or the JVM "
                  "crashed during startup.")
        log.error(f"  Last 10 non-debug lines of launcher.log:")
        lines = [line for line in content.splitlines()
                 if line and "DeadlockMonitor" not in line
                 and "AdManager" not in line]
        for line in lines[-10:]:
            # Gateway redacts the password in launcher.log, but account
            # numbers (DU/U) can appear; run through _redact_logs since
            # this fires on auth failure — exactly when users grab logs.
            log.error(f"    {_redact_logs(line)}")
        return

    if has_timeout and not has_ns_auth_start:
        log.error("Diagnosis: IBKR SILENTLY DROPPED THE AUTH REQUEST.")
        log.error("  Gateway reached 'Authenticating' and waited 20s for "
                  "IBKR to send NS_AUTH_START, but nothing came back. "
                  "This is typically one of:")
        log.error("    (1) IBKR rate-limit / cooldown after repeated "
                  "failed logins — wait 5-60 minutes and retry")
        log.error("    (2) Wrong regional server — check your TWS_SERVER "
                  "/ TWS_SERVER_PAPER value matches your account")
        log.error("    (3) Another session holding the account — "
                  "shut down any other container / mobile app / TWS "
                  "logged in as this user")
        log.error("  See controller/docs/BOOTSTRAP.md for more.")
        return

    if has_ns_auth_start and not has_timeout:
        # Not the bad-credentials signature: that one has a CCP Timeout!
        # after NS_AUTH_START (next branch). A field report on 2026-09-30
        # had an operator about to audit a correct password because this
        # message led with credentials; the same account then logged in
        # on its third plain retry.
        log.error(
            f"ALERT_LOGIN_FAILED mode={TRADING_MODE} "
            f"reason=\"post-auth-no-progress\" "
            f"suggested_action=\"IBKR answered the auth handshake but the "
            f"login never completed. This has cleared on a plain retry with "
            f"no configuration change, so retry before auditing "
            f"credentials; if it repeats, look for an unrecognized "
            f"post-auth dialog in the window dump\"")
        log.error("Diagnosis: auth request sent, server responded with "
                  "NS_AUTH_START, but we never reached PostAuthenticate.")
        log.error("  NS_AUTH_START means IBKR answered the handshake, so "
                  "this is not the bad-credentials signature — that one "
                  "times out after the handshake.")
        if _sibling_mode_ready():
            log.error("  The other trading mode is logged in on this "
                      "container, so the network path to IBKR works.")
        log.error("  Most likely causes, in order:")
        log.error("    (1) A transient stall on IBKR's side. The same "
                  "account has failed this way twice and then logged in "
                  "on the third try with nothing changed. Retry.")
        log.error("    (2) A post-auth dialog appeared that we didn't "
                  "recognize — check the window dump below")
        log.error("    (3) Only if it persists across several retries: "
                  "verify TWS_USERID / TWS_PASSWORD (or _PAPER variants)")
        return

    if has_ns_auth_start and has_timeout:
        log.error(
            f"ALERT_LOGIN_FAILED mode={TRADING_MODE} "
            f"reason=\"bad-credentials\" "
            f"suggested_action=\"IBKR rejected the credentials after "
            f"the handshake (NS_AUTH_START present, then timeout); "
            f"verify TWS_USERID / TWS_PASSWORD (or _PAPER variants) "
            f"and update env if password was rotated in IBKR Account "
            f"Settings\"")
        log.error("Diagnosis: auth request sent, server responded with "
                  "NS_AUTH_START, then auth timed out. Credentials were "
                  "probably rejected — verify TWS_USERID / TWS_PASSWORD.")
        return

    log.error("Diagnosis: unexpected state in launcher.log. Check "
              "the full launcher.log for details:")
    log.error(f"  docker exec <container> cat {launcher_log}")


def wait_for_api_port(timeout=180, answer_existing_session=True):
    """Probe Gateway's API port until it's actually accepting connections.

    This is the DEFINITIVE readiness signal — the API port only opens
    after Gateway has fully authenticated AND finished post-login setup.
    Swing-level "main window" markers proved unreliable because the
    main window shell is drawn even before login completes.

    Also opportunistically dismisses post-login disclaimer dialogs that
    appear during the wait — Gateway sometimes shows them right before
    opening the API port, and we want to clear them as they pop up.
    """
    api_port = api_port_for_mode()
    log.info(f"Probing API port {api_port} for readiness (up to {timeout}s)")
    start = time.monotonic()
    last_status = 0.0
    existing_session_handled = False
    while time.monotonic() - start < timeout:
        if is_api_port_open(api_port):
            elapsed = int(time.monotonic() - start)
            log.info(f"API port {api_port} accepting connections after {elapsed}s")
            return True
        # Issue #63: with TOTP, IBKR reports a competing session only after
        # the code is accepted, so "Existing session detected" can appear
        # here, after handle_2fa has returned. Answer it per
        # EXISTING_SESSION_DETECTED_ACTION, once per appearance (manual
        # leaves it up, and must not be re-handled every half second).
        try:
            windows = agent_windows() if answer_existing_session else []
        except Exception:
            windows = []
        if any(modal and ("existing session" in t.lower()
                          or "another session" in t.lower())
               for _w, t, modal in windows):
            if not existing_session_handled:
                existing_session_handled = True
                log.info("Existing-session dialog appeared while waiting for "
                         "the API port")
                if not handle_existing_session_dialog():
                    log.error("Existing-session dialog handling failed")
                time.sleep(0.5)
                continue
        else:
            existing_session_handled = False
            if windows:
                _note_existing_session_absent()

        # Opportunistically dismiss disclaimer dialogs each iteration.
        # Iterate SAFE_DISMISS_BUTTONS in order so BYPASS_WARNING-added
        # entries click deterministically, matching the order used by
        # dismiss_post_login_disclaimers().
        _, buttons = agent_list()
        for btn in SAFE_DISMISS_BUTTONS:
            if btn in buttons:
                log.info(f"  dismissing disclaimer {btn!r}")
                agent_click(btn)
                time.sleep(0.5)

        # Log progress every 10s with a snapshot of current windows
        now = time.monotonic()
        if now - last_status > 10:
            elapsed = int(now - start)
            windows = agent_windows()
            log.info(_redact_logs(
                f"  API port still closed at t+{elapsed}s; windows={windows}"))
            last_status = now
        time.sleep(0.5)
    return False


_OPERATOR_RESCUE_SECONDS = 300
# --- Upstream watchdog ---------------------------------------------------
# Field report 2026-10-03: Gateway lost its connection to IBKR on a Friday
# night (a "Connection to server failed ... SSLHandshakeException" modal)
# and never reconnected. The API port stayed open, so /health said healthy
# and API clients connected, but every request timed out for about 17 hours
# until someone restarted the container. Gateway shows the state in its main
# window: a label "Interactive Brokers API Server" followed by "connected".
#
# Rules, from the operator who hit it:
# - Measured on time.monotonic(): on a Docker VM it stops while the host
#   sleeps, so a long sleep does not read as a long outage.
# - The not-connected reading must hold on every poll for the whole grace
#   period; any other reading starts the count again.
# - upstream_connected / upstream_down_seconds always tell the truth. Only
#   the 503, ALERT_UPSTREAM_DOWN and the relaunch wait for the reset window
#   to end, and the grace period only counts time outside it.
# - During recovery /health says so (upstream_recovery_active,
#   next_relaunch_in_seconds), so an external monitor can defer instead of
#   restarting the container mid-backoff.
# - Before the first relaunch, the evidence is written to the log: a
#   relaunch destroys it, and nobody may be at the box.
# While recovery is active the watchdog alone decides what happens next: a
# relaunch that cannot log in during an IBKR outage leaves the port closed,
# and the port-closed path would otherwise escalate to the CCP halt.
_UPSTREAM_GRACE_SECONDS = int(os.environ.get("UPSTREAM_DOWN_GRACE_SECONDS", "600"))


def _upstream_relaunch_enabled():
    """Whether the watchdog may relaunch on its own. Default: only when
    the whole login runs unattended (a TOTP secret is set). With IB Key a
    relaunch means a phone prompt, possibly at 2 AM; with a passkey or a
    login finished over VNC it would end in a halt. Those users get the
    truthful 503 and ALERT_UPSTREAM_DOWN and decide themselves.
    UPSTREAM_RELAUNCH=yes / no overrides."""
    v = os.environ.get("UPSTREAM_RELAUNCH", "").strip().lower()
    if v in ("yes", "true", "1", "on"):
        return True
    if v in ("no", "false", "0", "off"):
        return False
    return bool(TOTP_SECRET)
_UPSTREAM_POLL_SECONDS = 30
_UPSTREAM_RELAUNCH_BACKOFF = (600, 1200, 2400, 3600)
_UPSTREAM_ANCHOR = "interactive brokers api server"

_upstream_connected = None       # last reading: True, False, or None (unknown)
_upstream_down_since = None      # monotonic; the truth, window or not
_upstream_grace_since = None     # monotonic; counts time outside the window only
_upstream_recovery = False
_upstream_next_relaunch_at = None
_upstream_relaunches = 0
_upstream_rejected_before = False   # last relaunch showed the bad-credentials fingerprint
_upstream_declared = False          # ALERT_UPSTREAM_DOWN raised for this outage
_upstream_dead_jvm = None           # the exited JVM already handled during recovery


def _read_upstream_connected():
    """True / False from Gateway's own status label, None when it cannot
    be read (no main window yet, agent unreachable, or not Gateway).

    The value is the label right after "Interactive Brokers API Server" in
    the same window. Matching "connected" anywhere would also hit the
    "API Client" count ("1 connected").
    """
    try:
        labels = agent_labels()
    except Exception:
        return None
    for i, (wtitle, text) in enumerate(labels):
        if text.strip().lower() != _UPSTREAM_ANCHOR:
            continue
        for other_title, value in labels[i + 1:]:
            if other_title == wtitle:
                # Only the two values seen in the field count. Anything
                # else (blank, a translation, a new wording) is unknown,
                # never a reason to relaunch.
                return {"connected": True, "disconnected": False}.get(
                    value.strip().lower())
        return None
    return None


def _upstream_reset(keep_backoff=False):
    """End the outage. ``keep_backoff`` after a relaunch that logged in
    but before Gateway has read connected: if it comes straight back, the
    next relaunches keep backing off instead of starting from 10 minutes."""
    global _upstream_down_since, _upstream_grace_since, _upstream_recovery
    global _upstream_next_relaunch_at, _upstream_relaunches
    global _upstream_rejected_before, _upstream_declared
    _upstream_declared = False
    _upstream_down_since = None
    _upstream_grace_since = None
    _upstream_recovery = False
    _upstream_next_relaunch_at = None
    _upstream_rejected_before = False
    if not keep_backoff:
        _upstream_relaunches = 0


def _upstream_health_fields():
    # Read each global once: the monitor loop may reset them between two
    # reads on the /health thread.
    now = time.monotonic()
    recovering, nxt_at = _upstream_recovery, _upstream_next_relaunch_at
    down_since = _upstream_down_since
    return {
        "upstream_connected": _upstream_connected,
        "upstream_down_seconds": (None if down_since is None
                                  else max(0.0, now - down_since)),
        "upstream_recovery_active": recovering,
        "next_relaunch_in_seconds": (max(0.0, nxt_at - now)
                                     if recovering and nxt_at is not None
                                     else None),
    }


def _launcher_log_tail(lines=30):
    path = os.path.join(JTS_CONFIG_DIR, "launcher.log")
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            f.seek(0, 2)
            f.seek(max(0, f.tell() - 65536))
            tail = [ln for ln in f.read().splitlines() if ln.strip()]
        return tail[-lines:]
    except OSError as e:
        return [f"(launcher.log unavailable: {type(e).__name__})"]


def _modal_sections(dump):
    """The lines of a WINDOW dump that belong to modal dialogs.

    WINDOW dumps every window whose title matches, and Gateway titles some
    error modals "IBKR Gateway", like its main frame, which comes first and
    is long (field reports 2026-10-03 and 10-04). Keeping only the modal
    sections stops the frame from crowding out the error text.
    """
    out, keep = [], False
    for ln in (dump or "").split("\n"):
        if ln.startswith("=== window="):
            keep = " modal=true" in ln
        if keep and ln and ln not in ("OK", "END"):
            out.append(ln)
    return out


def _capture_upstream_diagnostics():
    """Log what the operator would want to see before the relaunch erases
    it: the window list, every modal's text, the main-window labels, and
    the tail of this mode's launcher.log."""
    log.warning("Upstream diagnostics, captured before relaunching:")
    try:
        windows = agent_windows()
        log.warning(_redact_logs(f"  windows: {windows}"))
        for title in dict.fromkeys(t for _w, t, modal in windows if modal):
            for ln in _modal_sections(agent_window(title))[:60]:
                log.warning(_redact_logs(f"  modal [{title}] {ln}"))
    except Exception as e:
        log.warning(f"  windows: unavailable ({type(e).__name__}: {e})")
    try:
        for wtitle, text in agent_labels()[:60]:
            log.warning(_redact_logs(f"  label [{wtitle}] {text!r}"))
    except Exception as e:
        log.warning(f"  labels: unavailable ({type(e).__name__}: {e})")
    for ln in _launcher_log_tail():
        log.warning(_redact_logs(f"  launcher.log {ln}"))


def _upstream_tick(now=None):
    """Take one reading and advance the watchdog. True when a relaunch is
    due now."""
    global _upstream_connected, _upstream_down_since, _upstream_grace_since
    global _upstream_recovery, _upstream_next_relaunch_at, _upstream_declared
    now = time.monotonic() if now is None else now
    reading = _read_upstream_connected()
    _upstream_connected = reading
    if reading is True:
        if _upstream_declared:
            down = 0 if _upstream_down_since is None else now - _upstream_down_since
            log.info(f"UPSTREAM_RESTORED mode={TRADING_MODE} "
                     f"down_seconds={int(down)} relaunches={_upstream_relaunches}")
            if _current_state == State.UPSTREAM_DOWN:
                _set_state(State.MONITORING)
        elif _upstream_down_since is not None:
            # A blip that never became an outage, e.g. IBKR's nightly reset.
            log.info(f"Upstream: Gateway reconnected to IBKR after "
                     f"{int(now - _upstream_down_since)}s")
        _upstream_reset()
        return False
    if not _upstream_declared:
        if reading is None:
            # Not a reading of "disconnected": the run is broken.
            _upstream_down_since = None
            _upstream_grace_since = None
            return False
        if _upstream_down_since is None:
            _upstream_down_since = now
            log.info("Upstream: Gateway reports it is not connected to IBKR; "
                     "watching")
    if _is_ibkr_maintenance_window():
        _upstream_grace_since = None
        if _upstream_recovery and _upstream_next_relaunch_at is not None:
            # Give IBKR the same margin after the window as a new outage.
            _upstream_next_relaunch_at = max(_upstream_next_relaunch_at,
                                             now + _UPSTREAM_GRACE_SECONDS)
        return False
    if not _upstream_declared:
        if _upstream_grace_since is None:
            _upstream_grace_since = now
        if now - _upstream_grace_since < _UPSTREAM_GRACE_SECONDS:
            return False
        _upstream_declared = True
        relaunch = _upstream_relaunch_enabled()
        action = ("relaunching this mode; retries back off to hourly"
                  if relaunch else
                  "not relaunching, since logging in needs a person; restart "
                  "the container once IBKR is reachable (UPSTREAM_RELAUNCH=yes "
                  "lets the controller do it)")
        log.error(
            f"ALERT_UPSTREAM_DOWN mode={TRADING_MODE} "
            f"down_seconds={int(now - _upstream_down_since)} "
            f"reason=\"Gateway reports it is not connected to IBKR\" "
            f"action=\"{action}\"")
        _set_state(State.UPSTREAM_DOWN)
        _capture_upstream_diagnostics()
        if relaunch:
            _upstream_recovery = True
            _upstream_next_relaunch_at = now + (
                0 if _upstream_relaunches == 0
                else _UPSTREAM_RELAUNCH_BACKOFF[
                    min(_upstream_relaunches, len(_UPSTREAM_RELAUNCH_BACKOFF)) - 1])
    if not _upstream_recovery:
        return False
    return (_upstream_next_relaunch_at is not None
            and now >= _upstream_next_relaunch_at)


def _upstream_relaunch():
    """Relaunch this mode to reconnect it. On success recovery ends; on
    failure the next try waits 10, 20, 40, then at most 60 minutes. A
    failure only an operator can clear still halts."""
    global _upstream_relaunches, _upstream_next_relaunch_at, _upstream_connected
    global _upstream_rejected_before
    _upstream_relaunches += 1
    n = _upstream_relaunches
    log.warning(f"Upstream: relaunching this mode's Gateway to reconnect "
                f"(attempt {n})")
    streak_before = _ccp_lockout_streak
    ok = False
    try:
        ok = do_restart_in_place()
    except Exception as e:
        log.error(f"Upstream: relaunch raised {type(e).__name__}: {e}")
    if not ok and _twofa_needs_operator:
        if not _await_manual_login():
            _halt_for_operator(
                "2FA needs a change only an operator can make; see the "
                "ALERT_2FA_FAILED line above")
            return
        ok = True
    if ok:
        down = (0 if _upstream_down_since is None
                else time.monotonic() - _upstream_down_since)
        log.info(f"UPSTREAM_RESTORED mode={TRADING_MODE} "
                 f"down_seconds={int(down)} relaunches={n}")
        _upstream_reset(keep_backoff=True)
        _upstream_connected = None  # re-read on the next check
        _set_state(State.MONITORING)
        return
    if _ccp_lockout_streak > streak_before:
        # A CCP lockout is handled as one, with the operator's setting
        # (by default a halt): repeating relaunches against it is the
        # 2026-04-19 re-stranding pattern.
        log.error("Upstream: the relaunch hit a CCP lockout; handing over "
                  "to the CCP lockout handling")
        _upstream_reset()
        if _escalate_to_jvm_restart("CCP lockout on an upstream relaunch"):
            _set_state(State.MONITORING)
        return
    rejected_now = (_latest_login_rejected_credentials()
                    and not _is_ibkr_maintenance_window())
    if rejected_now and _upstream_rejected_before:
        log.error(
            f"ALERT_LOGIN_FAILED mode={TRADING_MODE} "
            f"reason=\"bad-credentials\" "
            f"suggested_action=\"two relaunches in a row were answered by "
            f"IBKR and then timed out; verify TWS_USERID / TWS_PASSWORD "
            f"(or _PAPER variants). If they are right, this was an "
            f"IBKR-side stall: restart the container\"")
        _halt_credentials_rejected(
            "two relaunches in a row were answered by IBKR and then timed "
            "out, the fingerprint of rejected credentials",
            if_right=". If the password is right, an IBKR-side stall "
                     "can look the same: just restart the container")
        return
    _upstream_rejected_before = rejected_now
    delay = _UPSTREAM_RELAUNCH_BACKOFF[min(n, len(_UPSTREAM_RELAUNCH_BACKOFF)) - 1]
    _upstream_next_relaunch_at = time.monotonic() + delay
    _set_state(State.UPSTREAM_DOWN)
    log.warning(f"Upstream: relaunch did not log in; next attempt in "
                f"{delay // 60} min")


# --- Logoff backstop ------------------------------------------------------
# Field reports 2026-10-01..03: Gateway's own Lock and Exit timer fired on
# only 1 of 5 observed days at 05:01 PM, across fresh and reused settings
# directories and three releases, with the value read back correctly each
# time. When the same JVM is still running 5 minutes after the configured
# logoff, log off on Gateway's behalf; the normal exit recovery logs in.
_LOGOFF_BACKSTOP_GRACE_SECONDS = 300
_JVM_STARTED_WALL = None
# Armed only when this JVM's own config step read back exactly
# AUTO_LOGOFF_TIME from Gateway, so the backstop only ever performs a
# logoff Gateway itself is scheduled to do. Not armed: Gateway in
# auto-restart mode (no logoff field), an AM value stored as PM, or a
# read-back that failed or couldn't run. Users who never set a logoff
# time are never logged off.
_logoff_backstop_armed = False
_logoff_backstop_armed_at = None   # when this session logged in (wall clock)
_LOGOFF_BACKSTOP_LATEST_SECONDS = 3600
_logoff_backstop_done_for = None
_logoff_backstop_warned = False


def _parse_wall_clock(value):
    """(hour, minute) in 24-hour time, or None.

    Accepts "05:01 PM" and 24-hour "17:01" / "00:30". A bare hour from 1
    to 12 is ambiguous in Gateway's 12-hour field, so it is refused rather
    than guessed. An AM value never arms the backstop anyway: Gateway
    stores it as PM, so the read-back fails.
    """
    m = re.fullmatch(r"\s*(\d{1,2}):(\d{2})\s*([AaPp][Mm])?\s*", value or "")
    if not m:
        return None
    hour, minute, meridiem = int(m.group(1)), int(m.group(2)), m.group(3)
    if minute > 59:
        return None
    if meridiem:
        if not 1 <= hour <= 12:
            return None
        return hour % 12 + (12 if meridiem.lower() == "pm" else 0), minute
    if hour == 0 or 13 <= hour <= 23:
        return hour, minute
    return None


def _note_session_login():
    """Record that the current session logged in now. The backstop only
    enforces a boundary this session was logged in before: a session that
    logged in at 17:25 is Gateway's to log off tomorrow, not now."""
    global _logoff_backstop_armed_at
    _logoff_backstop_armed_at = datetime.now()


def _gateway_zone_matches_local():
    """True when Gateway and this controller read the clock in the same
    zone, so the backstop's 17:01 is Gateway's 17:01.

    Gateway's zone is jts.ini's TimeZone when set (gnzsnz writes it from
    TIME_ZONE); otherwise the JVM's default, which comes from the same TZ
    this process uses. Nothing in the image ties TZ to TIME_ZONE, so a
    container with only TIME_ZONE set could run Gateway on New York time
    and this process on UTC — then the backstop stays off.
    """
    path = os.path.join(JTS_CONFIG_DIR, "jts.ini")
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            tz = next((ln.split("=", 1)[1].strip() for ln in f
                       if ln.startswith("TimeZone=")), "")
    except OSError:
        tz = ""
    if not tz:
        return True
    local = "unknown"
    try:
        now = datetime.now().astimezone()
        local = now.tzname()
        same = now.utcoffset() == now.astimezone(ZoneInfo(tz)).utcoffset()
    except Exception:
        same = False
    if not same:
        log.warning(f"Logoff backstop off: Gateway's time zone (jts.ini "
                    f"TimeZone={tz}) differs from this container's "
                    f"({local}). Set TZ to the same zone as TIME_ZONE to "
                    "enable it.")
    return same


def _logoff_backstop_boundary(now=None):
    """Today's logoff boundary when the backstop should act now, else None.

    Only for AUTO_LOGOFF_TIME without AUTO_RESTART_TIME, only after this
    JVM's read-back confirmed Gateway holds that exact AM/PM time in the
    same time zone, only for a session that logged in before the boundary
    (one that logged in at 17:25 is Gateway's to log off tomorrow), only
    between 5 and 60 minutes after it, and once per boundary.
    """
    global _logoff_backstop_warned
    value = os.environ.get("AUTO_LOGOFF_TIME", "").strip()
    if not value or os.environ.get("AUTO_RESTART_TIME", "").strip():
        return None
    if not _logoff_backstop_armed:
        return None
    # Only a time written with AM/PM, the way Gateway shows it: then the
    # read-back that armed the backstop matched Gateway's own summary
    # ("at 05:01 PM"), which reflects how Gateway read the value. A 24-hour
    # or bare value matches only the text typed into the field.
    hm = (_parse_wall_clock(value)
          if re.search(r"[AaPp][Mm]\s*$", value) else None)
    if hm is None:
        if not _logoff_backstop_warned:
            _logoff_backstop_warned = True
            log.warning(
                f"Logoff backstop off: write AUTO_LOGOFF_TIME the way Gateway "
                f"shows it, with AM or PM (05:01 PM); {value!r} can't be "
                "confirmed against Gateway's schedule.")
        return None
    now = datetime.now() if now is None else now
    today = now.replace(hour=hm[0], minute=hm[1], second=0, microsecond=0)
    # Yesterday's boundary too, so 11:55-11:59 PM still fires after
    # midnight. Only within an hour of the boundary: a host that slept
    # through it doesn't log off hours later.
    for boundary in (today, today - timedelta(days=1)):
        since = (now - boundary).total_seconds()
        if not (_LOGOFF_BACKSTOP_GRACE_SECONDS <= since
                < _LOGOFF_BACKSTOP_LATEST_SECONDS):
            continue
        if _logoff_backstop_armed_at is None or _logoff_backstop_armed_at >= boundary:
            return None
        if _logoff_backstop_done_for == boundary:
            return None
        return boundary
    return None


def _run_logoff_backstop(boundary):
    """Log off the way Gateway's own Lock and Exit does: a clean logout,
    then SIGTERM if that stalls. The JVM exit is picked up by the monitor
    loop, which recovers and logs in as it does after a scheduled logoff."""
    global _logoff_backstop_done_for
    _logoff_backstop_done_for = boundary
    started = (_JVM_STARTED_WALL.strftime("%Y-%m-%d %H:%M:%S")
               if _JVM_STARTED_WALL else "unknown")
    log.info(f"LOGOFF_BACKSTOP mode={TRADING_MODE} "
             f"boundary={boundary:%H:%M} jvm_started={started} "
             f"reason=\"Gateway's scheduled logoff did not happen; logging "
             f"off on its behalf\"")
    proc = GATEWAY_PROC
    pid = proc.pid if proc is not None else None
    ok, status, why = _attempt_clean_logout()
    log.info(f"ALERT_CLEAN_LOGOUT mode={TRADING_MODE} pid={pid} "
             f"status={status} reason=\"{why}\"")
    if ok or proc is None or proc.poll() is not None:
        return
    log.warning("LOGOFF_BACKSTOP: clean logout failed; sending SIGTERM")
    try:
        proc.terminate()
        proc.wait(timeout=30)
    except subprocess.TimeoutExpired:
        log.error("LOGOFF_BACKSTOP: JVM still alive 30s after SIGTERM; the "
                  "monitor loop keeps watching it")


_HALT_LOG_INTERVAL_SECONDS = 300


def _await_manual_login(timeout=_OPERATOR_RESCUE_SECONDS):
    """Give a human a window to finish a login the controller cannot.

    Deliberately the plain port probe: ``wait_for_api_port_with_retry``
    re-drives the login up to 8 times, which is the storm this path exists
    to avoid.
    """
    log.error(
        f"ALERT_2FA_MANUAL_WINDOW mode={TRADING_MODE} "
        f"timeout_seconds={timeout} "
        f"remediation=\"finish the login over VNC (port 5900) within "
        f"{timeout}s and the controller picks the session up; after that it "
        f"halts and makes no further attempts\"")
    log.error(f"Waiting up to {timeout}s for a manual login before halting. "
              "Connect to the container's VNC (port 5900) and finish it by "
              "hand if you want this session. Nothing is retried meanwhile.")
    # The operator is at the keyboard: leave any existing-session dialog
    # for them rather than clicking it under their cursor.
    return wait_for_api_port(timeout=timeout, answer_existing_session=False)


def _halt_for_operator(reason, cycles=None):
    """Stop working without exiting, and stay stopped.

    Exiting hands the container back to Docker's restart policy, which
    re-runs the same failing login every few minutes. Halting keeps the
    JVM, VNC and the health server up and waits for a person.

    CAVEAT (field report, 2026-09-27): /health answers 503 here, and the
    shipped HEALTHCHECK curls /health, so the container goes unhealthy.
    Plain Docker never restarts on unhealthy, so the halt holds — but
    anything that acts on health status (Kubernetes livenessProbe, Swarm,
    autoheal-style sidecars, a monitor with a "restart unhealthy" rule)
    will restart the container and recreate the login storm this exists to
    prevent. Point liveness at /ready, which stays 200 while the process is
    deliberately alive, and keep /health for readiness. README and
    docs/OBSERVABILITY.md say so too.

    ``cycles`` bounds the reminder loop for tests; None means forever.
    """
    _set_state(State.HALTED)
    log.error(
        f"ALERT_HALTED mode={TRADING_MODE} state=HALTED "
        f"reason=\"{reason}\"")
    log.error(f"HALTED: {reason}")
    log.error("No further login attempts will be made. Fix the cause and "
              "restart the container; VNC stays reachable if you want to "
              "finish this login by hand.")
    n = 0
    while cycles is None or n < cycles:
        time.sleep(_HALT_LOG_INTERVAL_SECONDS)
        n += 1
        log.error(f"HALTED: {reason} (waiting for an operator)")


# How many times a mode relaunches its own Gateway before giving up and
# handing recovery back to the container. Pauses double from 60 s.
_MODE_RECOVERY_ATTEMPTS = 3


def _recover_mode_in_place(reason, attempts=None):
    """Relaunch this mode's Gateway and log in again before giving up.

    Field report 2026-09-30: a live login failed with post-auth-no-progress,
    the controller exited, and in dual mode nothing restarted it for about
    nine hours. The same account then logged in on a plain retry. Exiting
    hands recovery to the container, which in dual mode disturbs the other
    mode too; healing here keeps the other mode untouched.

    Uses the monitor loop's proven relaunch (``do_restart_in_place``): it
    tears the JVM down and runs the whole pipeline, 2FA included. That is
    deliberately not ``attempt_inplace_relogin``, which only re-drives the
    login form and would stall at a TOTP prompt.

    Bounded, with pauses of 60, 120 and 240 s, so a failure that persists
    still reaches the container within about ten minutes instead of
    looping. A 2FA failure only an operator can clear is not retried: it
    takes the same rescue-window-then-halt path main() uses, because
    retrying it is the login storm v0.11.0 exists to prevent. Neither are
    rejected credentials: Gateway's own rejection dialog halts where it is
    seen, and two consecutive attempts with the launcher.log fingerprint
    halt here.
    """
    attempts = _MODE_RECOVERY_ATTEMPTS if attempts is None else attempts
    def rejected():
        # A stall is the likelier story inside IBKR's maintenance window.
        return (_latest_login_rejected_credentials()
                and not _is_ibkr_maintenance_window())
    # The attempt that brought us here.
    rejected_before = rejected()
    for attempt in range(1, attempts + 1):
        delay = min(60 * 2 ** (attempt - 1), 600)
        log.warning(f"Recovery: {reason}. Relaunching Gateway in {delay}s "
                    f"(attempt {attempt}/{attempts}) before handing this "
                    "mode back to the container.")
        time.sleep(delay)
        try:
            if do_restart_in_place():
                log.info("Recovery: this mode is back, without a container "
                         "restart")
                return True
        except Exception as e:
            log.error(f"Recovery: relaunch raised {type(e).__name__}: {e}")
        if _twofa_needs_operator:
            if _await_manual_login():
                log.info("Manual login detected — resuming the normal flow")
                return True
            _halt_for_operator(
                "2FA needs a change only an operator can make; see the "
                "ALERT_2FA_FAILED line above")
            return False
        rejected_now = rejected()
        if rejected_before and rejected_now:
            # Relaunched attempts run no diagnosis, so the alert monitors
            # page on may not have fired yet.
            log.error(
                f"ALERT_LOGIN_FAILED mode={TRADING_MODE} "
                f"reason=\"bad-credentials\" "
                f"suggested_action=\"two logins in a row were answered by "
                f"IBKR and then timed out; verify TWS_USERID / TWS_PASSWORD "
                f"(or _PAPER variants). If they are right, this was an "
                f"IBKR-side stall: restart the container\"")
            _halt_credentials_rejected(
                "two logins in a row were answered by IBKR and then timed "
                "out, the fingerprint of rejected credentials",
                if_right=". If the password is right, an IBKR-side stall "
                         "can look the same: just restart the container")
            return False
        rejected_before = rejected_now
    log.error(f"Recovery: {attempts} relaunches did not bring this mode "
              "back; exiting so the container can act")
    return False


def signal_ready():
    """Touch the readiness file so run.sh can start socat."""
    with open(READY_FILE, "w") as f:
        f.write(str(int(time.time())))
    log.info(f"Readiness signal: {READY_FILE}")


# ── Process lifecycle ───────────────────────────────────────────────────

gateway_proc = None  # global so signal handler can reach it


# v0.5.9: states with no CCP slot in flight. SIGTERM during these is
# safe — there's nothing for Gateway's close handler to release. Emit
# ALERT_CLEAN_LOGOUT status=safe_no_session for the grep pipeline so
# operators can confirm shutdown timing wasn't a factor when
# investigating subsequent lockouts.
_PRE_AUTH_STATES = {
    State.INIT, State.LAUNCHING, State.AGENT_WAIT,
    State.APP_DISCOVERY, State.LOGIN,
}

# v0.5.9: states where the main "IB Gateway" window exists and the
# v0.5.6 _attempt_clean_logout path should work. The intermediate
# post-login states (DISCLAIMERS, API_WAIT, CONFIG, READY,
# COMMAND_SERVER) are grouped with MONITORING because by that point
# Gateway has rendered its main shell window, so WINDOW_CLOSING has
# somewhere to land — even if a modal dialog (disclaimer, config) is
# visible on top.
_CLEAN_LOGOUT_ELIGIBLE_STATES = {
    State.DISCLAIMERS, State.API_WAIT, State.CONFIG,
    State.READY, State.COMMAND_SERVER, State.MONITORING,
}

# v0.5.9: 2FA dialog title substring. Matches ``TWOFA_WINDOW_SUBSTR``
# used in ``handle_2fa`` — kept separate so the shutdown path doesn't
# reach into a helper function's local constant.
_TWO_FA_WINDOW_TITLE_SUBSTR = "Second Factor"


def _classify_shutdown_for_state(state):
    """v0.5.9: decide the ALERT_CLEAN_LOGOUT status label for a SIGTERM
    received in ``state``, and whether to attempt the v0.5.6 UI-close
    path. Pure-logic helper so the decision table is unit-testable
    without running the signal handler.

    Returns ``(attempt_clean_logout, status_if_not_attempting, reason)``.
    If ``attempt_clean_logout`` is True, the caller delegates to
    ``_attempt_clean_logout`` and uses its returned status. Otherwise
    the caller emits ``status_if_not_attempting`` / ``reason`` directly
    and falls through to SIGTERM.

    Why dispatch here rather than always calling _attempt_clean_logout:
    the v0.5.6 helper looks for the main "IB Gateway" window. In
    pre-MONITORING states that window doesn't exist yet, so the helper
    always returns ``failed_unreachable`` — noisy and misleading.
    Worse, POST_LOGIN and TWO_FA have a CCP slot in flight but no
    main-window UI close path, so we need distinct status labels
    (``zombie_slot_cannot_release`` / ``cancelled_pending_2fa``) to
    tell operators that SIGTERM in those states is a slot-stranding
    event independent of Gateway responding.
    """
    if state in _PRE_AUTH_STATES:
        return (False, "safe_no_session",
                f"state={state.value}; no CCP slot held, SIGTERM is safe")
    if state == State.POST_LOGIN:
        return (False, "zombie_slot_cannot_release",
                f"state={state.value}; CCP slot in flight but Gateway has "
                "no main-window UI close path yet; SIGTERM will strand "
                "the slot server-side until IBKR's timeout drains it")
    if state == State.TWO_FA:
        # Caller tries agent_close_window on the 2FA dialog; if that
        # succeeds, Gateway cancels the half-authed handshake.
        return (True, "cancelled_pending_2fa",
                f"state={state.value}; attempting to close 2FA dialog to "
                "cancel the in-flight auth")
    if state in _CLEAN_LOGOUT_ELIGIBLE_STATES:
        return (True, "monitoring_ui_close",
                f"state={state.value}; attempting Gateway main-window close")
    # Unknown state — safest behaviour is to attempt clean logout
    # (matches v0.5.6) and let its result drive the status.
    return (True, "monitoring_ui_close",
            f"state={state.value} (unclassified); attempting clean logout")


def _attempt_state_aware_clean_logout(state):
    """v0.5.9: state-aware wrapper around _attempt_clean_logout.

    For MONITORING and post-auth pre-monitoring states: delegates to
    the v0.5.6 helper (find main window, post WINDOW_CLOSING, poll JVM
    exit). For TWO_FA: close the 2FA dialog first via the agent, then
    poll for JVM exit. Other states are handled upstream by
    ``_classify_shutdown_for_state`` and never reach this function.

    Returns ``(success, status, reason)`` mirroring
    ``_attempt_clean_logout``. Status values extended beyond v0.5.6:
    ``cancelled_pending_2fa`` (2FA dialog closed, JVM exited),
    ``failed_cancel_2fa`` (2FA dialog not found or agent rejected).
    """
    if state == State.TWO_FA:
        global GATEWAY_PROC
        if GATEWAY_PROC is None or GATEWAY_PROC.poll() is not None:
            return (True, "cancelled_pending_2fa", "JVM already exited")
        if not agent_close_window(_TWO_FA_WINDOW_TITLE_SUBSTR):
            return (False, "failed_cancel_2fa",
                    "agent CLOSE_WIN on 2FA dialog failed; "
                    "falling back to SIGTERM")
        deadline = time.monotonic() + _CLEAN_LOGOUT_TIMEOUT_SECONDS
        while time.monotonic() < deadline:
            if GATEWAY_PROC.poll() is not None:
                return (True, "cancelled_pending_2fa",
                        "2FA dialog closed; JVM exited cleanly within "
                        f"{_CLEAN_LOGOUT_TIMEOUT_SECONDS}s")
            time.sleep(0.25)
        return (False, "failed_cancel_2fa",
                "2FA dialog closed via agent but JVM did not exit within "
                f"{_CLEAN_LOGOUT_TIMEOUT_SECONDS}s; falling back to SIGTERM")
    return _attempt_clean_logout()


def shutdown(signum, frame):
    if signum == signal.SIGTERM:
        signame = "SIGTERM"
    elif signum == signal.SIGINT:
        signame = "SIGINT"
    else:
        signame = f"signal-{signum}"
    log.info(f"Received signal {signum} ({signame}), shutting down Gateway")
    # v0.5.6: read GATEWAY_PROC (uppercase) directly so post-restart
    # signals address the current JVM, not the original one. The
    # lowercase ``gateway_proc`` alias is only the module-load reference
    # and is not updated by ``_relaunch_and_login_in_place``.
    proc = GATEWAY_PROC if GATEWAY_PROC is not None else gateway_proc
    graceful = True
    clean_logout_applied = False
    # v0.5.9: state-aware shutdown. Each exit path emits
    # ALERT_CLEAN_LOGOUT with a distinct status so monitoring can
    # distinguish "safe, no slot held" from "slot in flight, can't
    # cleanly release".
    state = _current_state
    if proc is None or proc.poll() is not None:
        pid = "none"
        status = "safe_no_session"
        reason_text = f"no Gateway JVM present (state={state.value})"
        log.info(
            f"ALERT_CLEAN_LOGOUT mode={TRADING_MODE} pid={pid} "
            f"status={status} reason=\"{reason_text}\"")
    else:
        pid = proc.pid
        attempt_close, fallback_status, fallback_reason = (
            _classify_shutdown_for_state(state))
        if attempt_close:
            clean_success, clean_status, clean_reason = (
                _attempt_state_aware_clean_logout(state))
            log.info(
                f"ALERT_CLEAN_LOGOUT mode={TRADING_MODE} pid={pid} "
                f"status={clean_status} reason=\"{clean_reason}\"")
            clean_logout_applied = clean_success
        else:
            log.info(
                f"ALERT_CLEAN_LOGOUT mode={TRADING_MODE} pid={pid} "
                f"status={fallback_status} reason=\"{fallback_reason}\"")
        if not clean_logout_applied:
            proc.terminate()
            try:
                proc.wait(timeout=15)
            except subprocess.TimeoutExpired:
                log.warning("Gateway didn't terminate cleanly, killing")
                proc.kill()
                graceful = False
    try:
        os.unlink(READY_FILE)
    except FileNotFoundError:
        pass
    # ALERT_SHUTDOWN grep-contract token (v0.5.2). INFO-level so it doesn't
    # trip the ERROR-level wake-someone-up filters; monitors that want to
    # distinguish operator-initiated shutdowns from JVM crashes can grep on
    # the token itself. graceful=false means Gateway ignored SIGTERM and
    # had to be SIGKILL'd — worth flagging in dashboards as it points at
    # a JVM that's stuck (deadlocked Swing thread, blocked I/O, etc.).
    if clean_logout_applied:
        reason = (
            f"controller received {signame}; Gateway JVM exited via clean "
            "UI logout (WINDOW_CLOSING); no SIGTERM needed")
    elif graceful:
        reason = f"controller received {signame}; Gateway JVM exited cleanly within 15s"
    else:
        reason = (
            f"controller received {signame}; Gateway JVM did not exit within 15s "
            "of SIGTERM and was SIGKILL'd")
    log.info(
        f"ALERT_SHUTDOWN mode={TRADING_MODE} signal={signame} "
        f"graceful={'true' if graceful else 'false'} "
        f"reason=\"{reason}\"")
    sys.exit(0)


def _warn_unsupported_env_vars():
    """Warn loudly at startup about IBC env vars we don't implement.

    Users migrating from IBC may have these set in their .env without
    realizing the controller ignores them. A silent ignore means the
    user thinks they're getting a behavior they're not. This is the
    one chance to tell them at startup so they don't debug a
    phantom bug later.
    """
    # IBC env vars we still don't honor. Vars we DID wire up are:
    #   TWOFA_EXIT_INTERVAL / TWOFA_TIMEOUT_ACTION /
    #   RELOGIN_AFTER_TWOFA_TIMEOUT  → handle_2fa() respects all three
    #   BYPASS_WARNING  → _resolve_safe_dismiss_buttons() extends the
    #                     disclaimer allowlist at module load
    #   TWS_COLD_RESTART → apply_warm_state() skips when set
    # TWOFA_DEVICE is honored as of v0.7.0 (issue #7): handle_2fa()
    # reads Gateway's "Enter <method> code" prompt and refuses to type
    # the TOTP if the dialog is asking for a different method than
    # TWOFA_DEVICE names. Not warned.
    unsupported = {
        "CUSTOM_CONFIG":
            "not honored; the controller reads env vars directly and "
            "does not render an IBC config.ini",
    }
    hit = []
    for name, reason in unsupported.items():
        if os.environ.get(name):
            hit.append((name, reason))
    if hit:
        log.warning("These IBC-compat env vars are SET but NOT honored by "
                    "the controller. If you're migrating from IBC and "
                    "depending on any of these, you'll need to stay on IBC "
                    "for those specific behaviors:")
        for name, reason in hit:
            log.warning(f"  - {name}: {reason}")


def main():
    global gateway_proc

    # First line in the log: user-submitted logs previously carried no
    # version at all, so triage had to infer it from behaviour (field
    # report, 2026-09-27). Keep it cheap and greppable.
    log.info(f"ibg-controller v{__version__} starting "
             f"(mode={TRADING_MODE}, gateway={TWS_VERSION or 'unknown'}, "
             f"product={GATEWAY_OR_TWS})")

    if not USERNAME or not PASSWORD:
        log.error("TWS_USERID and TWS_PASSWORD must be set")
        sys.exit(2)

    _totp_problem = _validate_totp_secret(TOTP_SECRET)
    if _totp_problem:
        # Fail here, before any login, rather than crash with a traceback
        # at the 2FA dialog. Same ALERT prefix operators already watch;
        # the reason is new and additive.
        log.error(
            f"ALERT_2FA_FAILED mode={TRADING_MODE} "
            f"reason=\"TWOFACTOR_CODE is not a base32 secret\" "
            f"remediation=\"{_totp_problem}\"")
        sys.exit(2)

    _warn_unsupported_env_vars()

    signal.signal(signal.SIGTERM, shutdown)
    signal.signal(signal.SIGINT, shutdown)

    # Start the /health endpoint as the very first thing so monitoring
    # can see "controller alive but not yet in MONITORING state" during
    # a long login, and "no process bound" vs. "bound but 503" are
    # distinguishable from the outside.
    start_health_server()

    _set_state(State.LAUNCHING)
    # 1. Launch Gateway JVM
    global GATEWAY_PROC, CURRENT_APP, JVM_PID
    GATEWAY_PROC = launch_gateway()
    gateway_proc = GATEWAY_PROC  # local alias for readability below

    _set_state(State.AGENT_WAIT)
    # 2a. Wait for the input agent's Unix socket to come up. The agent is
    # loaded via -javaagent:, so it starts when the JVM starts — earlier
    # than the AT-SPI tree gets populated. If the jar is missing the agent
    # will never come up; the controller will then fail at first SETTEXT.
    if os.path.exists(AGENT_JAR):
        log.info(f"Waiting for input agent at {AGENT_SOCKET}")
        if not agent_wait_ready(timeout=60):
            log.error("Input agent never came up — text input will fail")
        else:
            log.info("Input agent is up")
            # Record the Gateway JVM PID so find_app() can distinguish our
            # Gateway from any other Gateway JVM running in the same
            # container. Critical for TRADING_MODE=both where both live
            # and paper JVMs appear as 'IBKR Gateway' in AT-SPI.
            JVM_PID = agent_get_pid()
            if JVM_PID is not None:
                log.info(f"Gateway JVM PID (from agent): {JVM_PID}")
            else:
                log.warning("Agent did not report PID — find_app will not "
                            "be able to disambiguate in dual-mode containers")

    _set_state(State.APP_DISCOVERY)
    # 2b. v0.5.12: AT-SPI desktop tree is no longer populated by Gateway
    # (the AtkWrapper bridge is disabled — see launch_gateway). find_app
    # returns a stub Accessible carrying the JVM PID, or None iff the
    # agent never reported a PID. Treat None as a fatal launch failure.
    log.info("Resolving Gateway app handle (agent-reported PID)")
    app = find_app(APP_NAME_CANDIDATES, timeout=120, match_pid=JVM_PID)
    if app is None:
        log.error("Gateway PID unknown (agent never reported one) — "
                  "cannot proceed without a JVM identity")
        sys.exit(1)
    CURRENT_APP = app
    log.info(f"App registered: {app.get_name()!r} (pid={JVM_PID})")

    # v0.5.10: cold-start maintenance-window guard. A container booting
    # inside IBKR's reset window (23:30-02:00 ET by default)
    # will drive Log In into a still-draining auth server; the click is
    # silently dropped → CCP LOCKOUT. Delay before the Log In click so
    # IBKR's teardown of any prior session on these credentials has time
    # to propagate. Same semantics as the mid-run recovery guard; see
    # docs/DISCONNECT_RECOVERY.md for the 2026-04-20/21 incident that
    # motivated this.
    if _is_ibkr_maintenance_window():
        _apply_maintenance_recovery_delay(
            "cold start inside IBKR maintenance window")

    _set_state(State.LOGIN)
    # 3. Drive the login dialog
    if not handle_login(app):
        log.error("Login dialog handling failed")
        sys.exit(1)

    if TEST_MODE:
        log.info("CONTROLLER_TEST_MODE=1 — waiting 5s for Gateway response then dumping window state")
        time.sleep(5)
        log.info("=== POST-CLICK WINDOW DUMP (via agent) ===")
        # v0.6.3: redact account numbers in titles; password fields are
        # already masked at the agent. This dump is opt-in (TEST_MODE)
        # but users still share its output, so keep it clean.
        try:
            for line in agent_window("").split("\n"):
                if line and line not in ("OK", "END"):
                    log.info(f"  {_redact_logs(line)}")
        except Exception as e:
            log.error(f"  agent_window dump failed: {type(e).__name__}: {e}")
        sys.exit(0)

    # 3a. Check for CCP lockout BEFORE entering the 2FA wait. Gateway's
    # auth timeout fires ~20s after the Log In click. If we detect it,
    # there's no point waiting 120s for a 2FA dialog — auth was rejected.
    #
    # Recovery is via IN-JVM relogin, never by killing the JVM. Killing
    # and relaunching creates a fresh auth handshake that IBKR treats
    # as a new session and keeps the CCP lockout armed. IBC's
    # ReloginAfterSecondFactorAuthenticationTimeout pattern stays
    # entirely within one JVM (same rationale). See v0.4.0 CHANGELOG.
    ccp_retry = 0
    while _detect_ccp_lockout(timeout=25):
        _apply_ccp_backoff()
        ccp_retry += 1
        if ccp_retry > _INPLACE_RELOGIN_MAX_ATTEMPTS:
            log.error(f"CCP lockout persists after {_INPLACE_RELOGIN_MAX_ATTEMPTS} "
                      "in-JVM relogin attempts; escalating to long-cool-down "
                      "JVM restart (v0.4.5)")
            _escalate_to_jvm_restart(
                f"{_INPLACE_RELOGIN_MAX_ATTEMPTS} in-JVM relogins exhausted in main CCP pre-loop")
            # Escalation returned True — fresh JVM is up, login completed,
            # API port open (do_restart_in_place handled it all). Rebind
            # app to the new JVM's AT-SPI reference and exit the CCP loop
            # — the post-login steps below are now redundant but
            # idempotent, so letting them run is harmless.
            app = CURRENT_APP
            break
        log.info(f"Retrying auth in-JVM after CCP backoff "
                 f"(attempt {ccp_retry}/{_INPLACE_RELOGIN_MAX_ATTEMPTS})")
        if not attempt_inplace_relogin(app):
            log.error("In-JVM relogin failed (login frame disposed or "
                      "handle_login failed); escalating to long-cool-down "
                      "JVM restart (v0.4.5)")
            _escalate_to_jvm_restart(
                "in-JVM relogin returned False in main CCP pre-loop")
            app = CURRENT_APP
            break
        # Loop back: _detect_ccp_lockout polls the NEW Log In click's
        # auth outcome. If no fresh Timeout! appears within 25s, the
        # lockout has cleared and we fall through to post-login.

    _set_state(State.POST_LOGIN)
    # 3b. Dismiss any modal dialog Gateway shows immediately after the
    # Log In click — most commonly the existing-session-detected dialog
    # if IBKR's session-tracker hasn't released the previous session.
    if not handle_post_login_dialogs(app):
        log.error("Post-login dialog handling failed")
        sys.exit(1)

    # Gate: _detect_ccp_lockout==False is ambiguous — it can mean "auth
    # progressed past CCP" OR "stuck in the connecting-to-server retry
    # loop, which doesn't emit the Timeout! signature". Only reset when
    # we're confident we've actually progressed. See v0.3.2 CHANGELOG.
    if not _detect_login_stuck_connecting():
        _reset_ccp_backoff()

    _set_state(State.TWO_FA)
    # 4. 2FA if applicable
    if not handle_2fa(app):
        log.error("2FA handling failed")
        if _twofa_needs_operator:
            # Needs a change only a human can make (the account's 2FA
            # methods, TWOFA_DEVICE, PASSKEY_AUTHENTICATE). Another
            # attempt fails identically, so offer a rescue window and then
            # halt rather than exiting into Docker's restart policy, which
            # would re-run this login every few minutes (issues #20, #37).
            if _await_manual_login():
                log.info("Manual login detected — resuming the normal flow")
            else:
                _halt_for_operator(
                    "2FA needs a change only an operator can make; see the "
                    "ALERT_2FA_FAILED line above")
        elif not _recover_mode_in_place("2FA handling failed"):
            sys.exit(1)

    _set_state(State.DISCLAIMERS)
    # 4b. Dismiss known post-login disclaimer dialogs (paper-trading
    # risks acknowledgement, terms-of-service updates, etc.) that
    # otherwise sit on screen forever. IBC normally auto-clicks these
    # via its BypassWarning settings.
    dismiss_post_login_disclaimers(timeout=30)

    _set_state(State.API_WAIT)
    # 5. Wait for the API port to actually be listening. This is the
    # definitive readiness signal — Swing-level "main window detected"
    # markers turned out to be unreliable (the buttons we look for exist
    # in Gateway's empty pre-login shell too). API port == authenticated.
    #
    # v0.4.1: wrapped in an in-JVM relogin retry loop so late-manifesting
    # stuck-connecting failures (which don't emit the launcher.log
    # ``Timeout!`` signature caught by the v0.4.0 outer CCP loop at
    # line 2383) are still recovered via attempt_inplace_relogin. A
    # terminal failure with no lockout signature still exits with the
    # same diagnostic dump as before.
    wait_for_api_port_with_retry(CURRENT_APP)

    _set_state(State.CONFIG)
    # 5b. Apply post-login API configuration (Master Client ID, Read-Only
    # API, etc.) if the user set any of the corresponding env vars. Runs
    # AFTER the API port is open so we know the Configure menu is real
    # and the settings dialog has live data. Non-fatal on failure — we
    # just warn and continue to readiness so the user gets a connectable
    # API even if the config knobs didn't take.
    handle_post_login_config()

    _set_state(State.READY)
    # 6. Signal ready
    signal_ready()

    _set_state(State.COMMAND_SERVER)
    # 6b. Optional IBC-compat command server. Listens on TCP port
    # (default 7462, IBC's default) for text commands like STOP /
    # RESTART / RECONNECTDATA / RECONNECTACCOUNT. Started in a
    # background thread so it runs alongside the monitor loop. Only
    # enabled if CONTROLLER_COMMAND_SERVER_PORT is set (defaults to off
    # so we don't grab a port users didn't ask for).
    start_command_server(app)

    _set_state(State.MONITORING)
    # 7. Long-running monitor: watch JVM + API port + re-auth events
    log.info("Login complete. Entering monitor loop.")
    monitor_loop(app)


# ── IBC-compat TCP command server (Phase 2.4) ─────────────────────────
#
# IBC ships a small TCP listener that accepts text commands from
# external orchestration tools (cron jobs, watchdogs). We implement
# the same protocol so existing users can swap in the controller
# without rewriting their orchestration.
#
# Protocol: line-based, plain text, one command per connection.
#   STOP              → clean shutdown of Gateway + controller
#   RESTART           → kill Gateway JVM and re-launch (NOT YET IMPLEMENTED)
#   RECONNECTDATA     → click File → Reconnect Data menu item
#   RECONNECTACCOUNT  → re-drive the full login flow via attempt_reauth
#   ENABLEAPI         → no-op (ApiOnly=true is already set in jts.ini)
#
# The listener runs in a daemon thread so it dies with the process.
# Single-connection at a time (no concurrency). Bind address is
# configurable via CONTROLLER_COMMAND_SERVER_HOST (default 0.0.0.0, so a
# published port reaches it). Set CONTROLLER_COMMAND_SERVER_AUTH_TOKEN:
# -p 127.0.0.1:... keeps the host's network out, but not other
# containers on the same Docker network.

_command_server_thread = None
_command_server_app = None
_command_server_socket = None


def start_command_server(app):
    """Start the IBC-compat command server in a daemon thread if
    CONTROLLER_COMMAND_SERVER_PORT is set. No-op otherwise."""
    global _command_server_thread, _command_server_app
    port_str = os.environ.get("CONTROLLER_COMMAND_SERVER_PORT", "").strip()
    if not port_str:
        log.info("Command server: CONTROLLER_COMMAND_SERVER_PORT not set, skipping")
        return
    try:
        port = int(port_str)
    except ValueError:
        log.warning(f"Command server: invalid port {port_str!r}, skipping")
        return
    # Default to 0.0.0.0 because the controller is Docker-first — the
    # container's localhost is not reachable from the host's `docker
    # run -p 7462:7462`, you need to bind to the container's external
    # interface. Users who want to restrict exposure should rely on
    # Docker's port mapping (e.g. -p 127.0.0.1:7462:7462) rather than
    # the in-container bind address.
    host = os.environ.get("CONTROLLER_COMMAND_SERVER_HOST", "0.0.0.0").strip()

    _command_server_app = app
    _command_server_thread = threading.Thread(
        target=_command_server_main,
        args=(host, port),
        daemon=True,
        name="command-server",
    )
    _command_server_thread.start()
    log.info(f"Command server: listening on {host}:{port}")


def _command_server_main(host, port):
    """Accept loop for the IBC-compat command server. Runs forever
    until the socket is closed (on process exit via daemon-thread
    teardown).

    If CONTROLLER_COMMAND_SERVER_AUTH_TOKEN is set, clients must send
    an AUTH line before any command, e.g.:
        AUTH <token>
        STOP
    The token is checked with hmac.compare_digest to resist timing
    side channels. Connections that don't AUTH, or send a wrong token,
    get an ERR and are closed without reaching _handle_command.
    Connections with no token configured run as before (IBC behavior).
    """
    global _command_server_socket
    auth_token = os.environ.get("CONTROLLER_COMMAND_SERVER_AUTH_TOKEN", "")
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen(5)
        _command_server_socket = s
        if auth_token:
            log.info("Command server: auth token required")
        else:
            log.warning("Command server: NO AUTH TOKEN SET. Anyone "
                        "who can reach the bind address can send STOP / "
                        "RESTART / RECONNECTACCOUNT. For localhost-only "
                        "access use docker -p 127.0.0.1:PORT:PORT; for "
                        "remote access set CONTROLLER_COMMAND_SERVER_AUTH_TOKEN.")
    except Exception as e:
        log.error(f"Command server: bind/listen failed: {type(e).__name__}: {e}")
        return

    # Simple rate limit: cap how fast we can accept new connections.
    # Protects against someone spamming the server even with a valid
    # token (defense in depth — a real brute-force attacker would
    # grind through AUTH attempts faster than this allows).
    min_accept_interval = 0.25  # 4 accepts/sec max
    last_accept = 0.0

    while True:
        try:
            conn, addr = s.accept()
        except Exception as e:
            log.warning(f"Command server: accept error: {type(e).__name__}: {e}")
            return

        now = time.monotonic()
        delay = min_accept_interval - (now - last_accept)
        if delay > 0:
            time.sleep(delay)
        last_accept = time.monotonic()

        try:
            conn.settimeout(5.0)

            # Read up to two lines: optional AUTH <token>, then the
            # command. Max payload kept small to avoid buffer-bloat
            # attacks.
            data = b""
            while data.count(b"\n") < 2 and len(data) < 1024:
                chunk = conn.recv(256)
                if not chunk:
                    break
                data += chunk
                if auth_token == "" and b"\n" in data:
                    break

            text = data.decode("utf-8", errors="replace")
            lines = text.splitlines()
            cmd = ""

            if auth_token:
                # Expect: first line AUTH <token>, second line <command>
                if not lines or not lines[0].upper().startswith("AUTH "):
                    conn.sendall(b"ERR auth_required\n")
                    log.warning(f"Command server: {addr[0]}:{addr[1]} rejected — no AUTH line")
                    continue
                provided = lines[0][5:].strip()
                if not hmac.compare_digest(provided, auth_token):
                    conn.sendall(b"ERR auth_failed\n")
                    log.warning(f"Command server: {addr[0]}:{addr[1]} rejected — auth failed")
                    continue
                if len(lines) < 2:
                    conn.sendall(b"ERR empty_command\n")
                    continue
                cmd = lines[1].strip().upper()
            else:
                cmd = (lines[0].strip().upper() if lines else "")

            log.info(f"Command server: {addr[0]}:{addr[1]} sent {cmd!r}")
            response = _handle_command(cmd)
            try:
                conn.sendall((response + "\n").encode("utf-8"))
            except Exception:
                pass
        except Exception as e:
            log.warning(f"Command server: handler error: {type(e).__name__}: {e}")
        finally:
            try:
                conn.close()
            except Exception:
                pass


# ── HTTP health endpoint (v0.4.9) ──────────────────────────────────────
#
# Separate from the TCP command server on purpose: monitoring tools
# (Docker HEALTHCHECK, uptime checkers, Prometheus blackbox_exporter)
# want to curl a URL, not speak the IBC text protocol. Keeping them
# separate also means the command server can stay auth-gated without
# forcing monitoring to carry a token.
#
# Protocol: GET /health → 200 with JSON if state==MONITORING and the
# API port is open and the JVM process is alive, 503 with the same JSON
# otherwise. Any other path or method → 404. The body is always JSON so
# clients can parse it either way.
#
# Port selection mirrors the command server's dual-mode offset: paper
# gets base+1 in docker/run.sh so both controllers can bind on the same
# container with a single env var.

_health_server_thread = None
_health_server_httpd = None


def _socat_port():
    """The published port run.sh forwards to this mode's API port.

    run.sh's set_ports exports SOCAT_PORT before starting the controller
    (4003 live, 4004 paper on Gateway). None outside the image, where
    nothing forwards.
    """
    try:
        return int(os.environ.get("SOCAT_PORT", "") or 0) or None
    except ValueError:
        return None


def _build_health_snapshot():
    """Return a dict describing the controller's current health. Pure
    read of module globals — safe to call from any thread. Does NOT
    take locks; individual values may be racy but the overall shape is
    consistent enough for monitoring purposes."""
    api_port = api_port_for_mode()
    api_open = False
    try:
        api_open = is_api_port_open(api_port)
    except Exception:
        api_open = False

    # The port clients actually use. A connect here tests socat's LISTENER,
    # not the path through to Gateway — socat accepts and only then dials
    # 127.0.0.1:<api_port> — so this answers "is the forwarder alive", and
    # api_port_open answers "is Gateway serving". Both are needed: a field
    # report on 2026-09-27 had Gateway healthy and every client cut off for
    # about 29 hours because the forwarder was gone. Not counted into
    # "status": run.sh starts socat a moment after readiness, so counting it
    # would flap every login, and run.sh now restarts a dead forwarder.
    socat_port = _socat_port()
    socat_open = None
    if socat_port:
        try:
            socat_open = is_api_port_open(socat_port)
        except Exception:
            socat_open = False

    jvm_pid = JVM_PID
    jvm_alive = False
    if GATEWAY_PROC is not None:
        try:
            jvm_alive = GATEWAY_PROC.poll() is None
        except Exception:
            jvm_alive = False

    now = time.time()
    last_auth_ts = _last_auth_success_ts
    last_auth_age = None
    if last_auth_ts is not None:
        last_auth_age = max(0.0, now - last_auth_ts)

    state_name = _current_state.value if _current_state is not None else "UNKNOWN"
    healthy = (
        state_name == State.MONITORING.value
        and api_open
        and jvm_alive
    )

    return {
        "status": "healthy" if healthy else "unhealthy",
        "version": __version__,
        "mode": TRADING_MODE,
        "state": state_name,
        "jvm_pid": jvm_pid,
        "jvm_alive": jvm_alive,
        "api_port": api_port,
        "api_port_open": api_open,
        "socat_port": socat_port,
        "socat_port_open": socat_open,
        "last_auth_success_ts": last_auth_ts,
        "last_auth_success_age_seconds": last_auth_age,
        "ccp_lockout_streak": _ccp_lockout_streak,
        "ccp_backoff_seconds": _ccp_backoff_seconds,
        "uptime_seconds": max(0.0, now - _CONTROLLER_START_TS),
        **_upstream_health_fields(),
    }


class _HealthHandler(BaseHTTPRequestHandler):
    """Minimal GET /health handler. Also serves /ready as a shallower
    probe (always 200 as long as the process is running — useful for
    Kubernetes-style readiness where "process up" is the signal).
    """

    timeout = 5  # seconds an idle connection may hold its thread

    def log_message(self, format, *args):
        # Silence the default stderr access log. The controller's own
        # logger stays the single source of truth, and Docker
        # HEALTHCHECK would otherwise spam stderr every 30s.
        return

    def do_GET(self):
        path = self.path.split("?", 1)[0].rstrip("/") or "/"
        if path == "/health":
            snapshot = _build_health_snapshot()
            body = json.dumps(snapshot).encode("utf-8")
            status = 200 if snapshot["status"] == "healthy" else 503
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(body)
            return
        if path == "/ready":
            body = b'{"status":"up"}'
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        self.send_response(404)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"error":"not_found"}')


def start_health_server():
    """Start the /health HTTP server in a daemon thread if
    CONTROLLER_HEALTH_SERVER_PORT is set. No-op otherwise. Safe to call
    more than once — idempotent on the module thread global."""
    global _health_server_thread, _health_server_httpd
    if _health_server_thread is not None and _health_server_thread.is_alive():
        return
    port_str = os.environ.get("CONTROLLER_HEALTH_SERVER_PORT", "").strip()
    if not port_str:
        log.info("Health server: CONTROLLER_HEALTH_SERVER_PORT not set, skipping")
        return
    try:
        port = int(port_str)
    except ValueError:
        log.warning(f"Health server: invalid port {port_str!r}, skipping")
        return
    host = os.environ.get("CONTROLLER_HEALTH_SERVER_HOST", "0.0.0.0").strip()

    try:
        # Threaded, with a per-connection timeout: on a single-threaded
        # server one idle connection blocked /health and /ready for
        # everyone, failing the healthcheck and any liveness probe.
        httpd = ThreadingHTTPServer((host, port), _HealthHandler)
        httpd.daemon_threads = True
    except Exception as e:
        log.error(f"Health server: bind/listen failed: {type(e).__name__}: {e}")
        return
    _health_server_httpd = httpd

    def _serve():
        try:
            httpd.serve_forever(poll_interval=0.5)
        except Exception as e:
            log.warning(f"Health server: serve_forever error: {type(e).__name__}: {e}")

    _health_server_thread = threading.Thread(
        target=_serve, daemon=True, name="health-server")
    _health_server_thread.start()
    log.info(f"Health server: listening on {host}:{port} (GET /health, /ready)")


# v0.5.5: SIGTERM grace period extended from 20s to 30s. Gateway's JVM
# shutdown hooks need time to close the CCP session cleanly on IBKR's
# side; a too-short grace forces SIGKILL, which leaves IBKR holding the
# session slot until its own timeout fires. Env override for environments
# where even 30s isn't enough.
_JVM_TEARDOWN_GRACE_SECONDS = int(os.environ.get("JVM_TEARDOWN_GRACE_SECONDS", "30"))

# v0.5.6: Matches the titles of the Gateway main frame across the
# versions observed in testing ("IB Gateway", "IBKR IB Gateway").
# The agent's findWindowByTitleSubstring() takes a substring match,
# so any stable fragment works.
# 2026-04-27: widened from "IB Gateway" to "Gateway" — observed window
# title in 10.45.1c is "IBKR Gateway" (no space after IB), so the prior
# substring missed it and clean logout fell through to SIGTERM. "Gateway"
# is the stable fragment across all known title variants and is not a
# substring of any other top-level window title (auth dialogs are
# "Authenticating...", "Second Factor Authentication", etc.).
_GATEWAY_MAIN_WINDOW_TITLE_SUBSTR = "Gateway"

# v0.5.6: How long to wait for the JVM to exit cleanly after driving a
# WINDOW_CLOSING event to Gateway's main frame. Gateway's registered
# WindowListener performs a proper CCP session-close which can take a
# few seconds (network round-trip to IBKR + state flush) before the
# JVM terminates. If this expires, _teardown_jvm_for_restart falls
# through to the v0.5.5 SIGTERM path. Shorten in containers where
# Docker's stop-grace-period is tight; lengthen on slow networks.
_CLEAN_LOGOUT_TIMEOUT_SECONDS = int(
    os.environ.get("CLEAN_LOGOUT_TIMEOUT_SECONDS", "15"))


def _attempt_clean_logout(timeout_seconds=None):
    """v0.5.6: Drive Gateway to close cleanly via its UI close handler
    before we resort to SIGTERM. Returns ``(success, status, reason)``
    where ``status`` is one of:

    - ``"succeeded"`` — JVM exited within ``timeout_seconds`` of the
      WINDOW_CLOSING dispatch. Gateway's close handler ran, so the CCP
      session slot was cleanly released on IBKR's side. No stranded
      slot, no need for the adaptive cool-down to absorb anything.
    - ``"failed_unreachable"`` — the agent didn't respond to CLOSE_WIN
      (socket missing, EDT deadlocked before we could post the event,
      agent not yet initialised). Caller should fall through to the
      SIGTERM path.
    - ``"failed_timeout"`` — the agent accepted CLOSE_WIN but the JVM
      didn't exit within ``timeout_seconds``. Gateway's WindowListener
      is stuck; caller falls through to SIGTERM, and if *that* also
      times out the v0.5.5 adaptive cool-down absorbs the stranded
      slot.

    Callers must have already checked that ``GATEWAY_PROC`` is alive.
    This function tolerates the JVM exiting on its own mid-call (race
    between the outer check and here) and reports success.
    """
    if timeout_seconds is None:
        timeout_seconds = _CLEAN_LOGOUT_TIMEOUT_SECONDS

    global GATEWAY_PROC
    if GATEWAY_PROC is None or GATEWAY_PROC.poll() is not None:
        return (True, "succeeded", "JVM already exited")

    if not agent_close_window(_GATEWAY_MAIN_WINDOW_TITLE_SUBSTR):
        return (False, "failed_unreachable",
                "agent CLOSE_WIN did not succeed; falling back to SIGTERM")

    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        if GATEWAY_PROC.poll() is not None:
            return (True, "succeeded",
                    f"JVM exited cleanly within {timeout_seconds}s of "
                    f"WINDOW_CLOSING")
        time.sleep(0.25)

    return (False, "failed_timeout",
            f"JVM still alive {timeout_seconds}s after WINDOW_CLOSING "
            f"dispatched; Gateway close handler may be stalled")


def _teardown_jvm_for_restart():
    """v0.4.6: Terminate the current Gateway JVM and clear per-instance
    state files (agent socket, ready file) so a subsequent launch starts
    from a clean slate.

    Split out of ``do_restart_in_place`` so ``_escalate_to_jvm_restart``
    can kill the JVM *before* the long CCP cool-down. When the JVM is
    kept alive during the cool-down its internal "Attempt N: connecting
    to server" retry loop keeps banging on IBKR's auth server, which
    keeps the CCP rate limiter armed — observed 2026-04-16 with v0.4.5
    (20min cool-down with JVM alive → CCP still locked on restart).
    Killing the JVM first makes the cool-down genuinely silent from
    IBKR's perspective, which is the whole point.

    v0.5.5: extended SIGTERM grace from 20s to 30s so Gateway's JVM
    shutdown hooks have time to send a clean CCP session-close to IBKR
    before fallback SIGKILL. When SIGKILL is required, emits
    ``ALERT_JVM_UNCLEAN_SHUTDOWN`` — the suspected root cause of the
    persistent-lockout pattern where consecutive CCP lockouts accumulate
    faster than IBKR's session-slot timeout can drain them.

    v0.5.6: before SIGTERM, attempt a UI-driven clean logout by posting
    a WINDOW_CLOSING event to Gateway's main frame via the agent. This
    fires Gateway's registered WindowListener (the same code path as a
    user clicking the window's X button), which performs a proper CCP
    session-close before the JVM exits. If clean logout succeeds, no
    SIGTERM is needed and no slot is stranded — the root cause of the
    v0.5.5 incident is eliminated. If it fails (agent unreachable, EDT
    stalled, WindowListener itself broken), falls through to the v0.5.5
    SIGTERM + adaptive-cool-down defence.
    """
    global GATEWAY_PROC
    if GATEWAY_PROC is not None and GATEWAY_PROC.poll() is None:
        pid = GATEWAY_PROC.pid
        log.info(f"RESTART: terminating Gateway PID {pid}")

        # v0.5.6: try clean UI logout first. ALERT_CLEAN_LOGOUT is part
        # of the public stability contract — monitoring can grep for it
        # to track clean-logout success rate over time.
        clean_success, clean_status, clean_reason = _attempt_clean_logout()
        log.info(
            f"ALERT_CLEAN_LOGOUT mode={TRADING_MODE} pid={pid} "
            f"status={clean_status} reason=\"{clean_reason}\"")

        if not clean_success:
            clean = True
            unclean_reason = ""
            try:
                GATEWAY_PROC.terminate()
                try:
                    GATEWAY_PROC.wait(timeout=_JVM_TEARDOWN_GRACE_SECONDS)
                except subprocess.TimeoutExpired:
                    log.warning(
                        f"RESTART: Gateway didn't exit within "
                        f"{_JVM_TEARDOWN_GRACE_SECONDS}s grace; SIGKILL'ing. "
                        "IBKR session slot may be held server-side until timeout.")
                    GATEWAY_PROC.kill()
                    GATEWAY_PROC.wait(timeout=5)
                    clean = False
                    unclean_reason = (
                        f"Gateway JVM ignored SIGTERM within "
                        f"{_JVM_TEARDOWN_GRACE_SECONDS}s grace; required SIGKILL")
            except Exception as e:
                log.warning(f"RESTART: Gateway termination error: {e}")
                clean = False
                unclean_reason = f"teardown raised {type(e).__name__}: {e}"
            if not clean:
                # Stable grep token for external monitoring. Emitted once per
                # unclean teardown, distinct from ALERT_SHUTDOWN (which is
                # the lifecycle signal for controller exit). See
                # docs/OBSERVABILITY.md for the grep-contract guarantees.
                log.warning(
                    f"ALERT_JVM_UNCLEAN_SHUTDOWN mode={TRADING_MODE} "
                    f"pid={pid} reason=\"{unclean_reason}\" "
                    f"implication=\"IBKR CCP session slot likely held "
                    f"server-side until timeout; next auth attempt may hit "
                    f"lockout despite cool-down\"")
    GATEWAY_PROC = None

    for p in (AGENT_SOCKET, READY_FILE):
        try:
            os.unlink(p)
        except FileNotFoundError:
            pass
        except Exception as e:
            log.warning(f"RESTART: couldn't remove {p}: {e}")


def _relaunch_and_login_in_place():
    """v0.4.6: Launch a fresh Gateway JVM and re-run the full login
    pipeline. Assumes ``_teardown_jvm_for_restart`` has already run —
    i.e., GATEWAY_PROC is None, agent socket is cleared.

    Returns True on full success (re-login completed, API port open,
    readiness re-signalled), False on any failure. Updates the module
    globals GATEWAY_PROC / CURRENT_APP / JVM_PID so the monitor loop
    picks up the new references on its next iteration.
    """
    global GATEWAY_PROC, CURRENT_APP, JVM_PID, _command_server_app

    GATEWAY_PROC = launch_gateway()

    # 4. Re-discover the agent (new JVM = new socket).
    if not agent_wait_ready(timeout=60):
        log.error("RESTART: agent did not come up after re-launch")
        return False
    JVM_PID = agent_get_pid()
    log.info(f"RESTART: new Gateway JVM PID (from agent): {JVM_PID}")

    # 5. Re-discover the app in the AT-SPI tree.
    new_app = find_app(APP_NAME_CANDIDATES, timeout=120, match_pid=JVM_PID)
    if new_app is None:
        log.error("RESTART: Gateway app never reappeared in AT-SPI tree")
        return False
    CURRENT_APP = new_app
    _command_server_app = new_app  # the command server now drives the new app

    # 6. Re-drive login through the same pipeline main() uses.
    if not handle_login(new_app):
        log.error("RESTART: login dialog handling failed")
        return False

    # Check for CCP lockout on the freshly-relaunched JVM. If detected,
    # recovery must be IN-JVM — do NOT recurse do_restart_in_place here.
    # Killing this JVM and spawning another one is exactly what keeps
    # IBKR's CCP limiter armed (each new JVM = new auth handshake).
    # Stay in this JVM and re-click Log In instead. See v0.4.0 CHANGELOG.
    ccp_retry = 0
    while _detect_ccp_lockout(timeout=25):
        _apply_ccp_backoff()
        ccp_retry += 1
        if ccp_retry > _INPLACE_RELOGIN_MAX_ATTEMPTS:
            log.error(f"RESTART: CCP lockout persists after "
                      f"{_INPLACE_RELOGIN_MAX_ATTEMPTS} in-JVM relogin "
                      "attempts on the relaunched JVM; giving up")
            return False
        log.info(f"RESTART: in-JVM relogin after CCP backoff "
                 f"(attempt {ccp_retry}/{_INPLACE_RELOGIN_MAX_ATTEMPTS})")
        if not attempt_inplace_relogin(new_app):
            log.error("RESTART: in-JVM relogin failed on the relaunched JVM")
            return False

    # Gate: same reasoning as the main() path — don't reset on the
    # stuck-connecting case, only on genuine CCP-gate progress.
    if not _detect_login_stuck_connecting():
        _reset_ccp_backoff()

    if not handle_post_login_dialogs(new_app):
        log.error("RESTART: post-login dialog handling failed")
        return False
    if not handle_2fa(new_app):
        log.error("RESTART: 2FA handling failed")
        return False
    dismiss_post_login_disclaimers(timeout=30)
    if not wait_for_api_port(timeout=180):
        log.error("RESTART: API port never opened after re-launch")
        return False

    # 7. Re-apply post-login configuration (Master Client ID, etc.).
    handle_post_login_config()

    # 8. Re-signal readiness so run.sh / external watchers see the
    #    second round has stabilized.
    signal_ready()

    log.info("RESTART: Gateway re-launched successfully")
    return True


def do_restart_in_place():
    """Tear down the current Gateway JVM, clear per-instance state
    files, and re-run the full login sequence. Updates the module
    globals GATEWAY_PROC / CURRENT_APP / JVM_PID so the monitor loop
    picks up the new references on its next iteration.

    Used by the IBC-compat RESTART command and the monitor-loop wedge
    escalation. Returns True on full success (re-login completed, API
    port open, readiness re-signalled), False on any failure during
    the re-launch sequence.

    v0.4.6: Thin wrapper over ``_teardown_jvm_for_restart`` +
    ``_relaunch_and_login_in_place`` so ``_escalate_to_jvm_restart`` can
    interleave a long CCP cool-down *between* the teardown and the
    relaunch (JVM dead during the cool-down = genuinely silent from
    IBKR's perspective = CCP limiter actually clears).
    """
    with _restart_lock:
        _teardown_jvm_for_restart()
        return _relaunch_and_login_in_place()


# One relaunch at a time: the command server's RESTART, the monitor loop's
# recovery and the upstream watchdog could otherwise overlap. Reentrant,
# because TWOFA_TIMEOUT_ACTION=restart relaunches from inside a relaunch.
_restart_lock = threading.RLock()


def _relaunch_in_progress():
    """True when another thread holds the relaunch lock."""
    if _restart_lock.acquire(blocking=False):
        _restart_lock.release()
        return False
    return True


def _handle_command(cmd):
    """Dispatch a single command string to the appropriate controller
    action. Returns the response string to send back to the client."""
    if cmd in ("RESTART", "RECONNECTACCOUNT"):
        if _current_state == State.HALTED:
            # A halt means a login must not be retried until a person
            # fixes the cause; a login now would repeat what it stopped.
            return (f"ERR {cmd}: halted; fix the cause and restart the "
                    "container")
        if _relaunch_in_progress():
            return f"ERR {cmd}: a relaunch is already in progress"
    if cmd == "STOP":
        log.info("Command server: STOP received — initiating shutdown")
        # Send SIGTERM to ourselves so the existing shutdown path runs
        try:
            os.kill(os.getpid(), signal.SIGTERM)
        except Exception as e:
            return f"ERR STOP: {type(e).__name__}: {e}"
        return "OK STOP"

    if cmd == "RESTART":
        log.info("Command server: RESTART — tearing down and re-launching Gateway")
        try:
            ok = do_restart_in_place()
        except Exception as e:
            return f"ERR RESTART: {type(e).__name__}: {e}"
        return "OK RESTART" if ok else "ERR RESTART: re-launch failed"

    if cmd == "RECONNECTDATA":
        # Gateway 10.45's File menu has: Gateway Logs, API Logs, Gateway
        # Layout/Settings, Close. No "Reconnect Data" item. RECONNECTDATA
        # is a TWS-only command (on TWS the File menu has Reconnect Data
        # and Reconnect Account). On Gateway, the equivalent workflow is
        # to RECONNECTACCOUNT (re-login) instead.
        product = os.environ.get("GATEWAY_OR_TWS", "gateway").strip().lower()
        if product == "tws":
            log.info("Command server: RECONNECTDATA — clicking File → Reconnect Data")
            if not agent_click("File"):
                return "ERR RECONNECTDATA: could not click File menu"
            time.sleep(0.3)
            if not agent_click("Reconnect Data"):
                return "ERR RECONNECTDATA: could not click Reconnect Data item"
            return "OK RECONNECTDATA"
        return ("ERR RECONNECTDATA: not supported on Gateway — "
                "Gateway's File menu has no Reconnect Data item. "
                "Use RECONNECTACCOUNT to re-drive the login flow instead.")

    if cmd == "RECONNECTACCOUNT":
        # Re-drive the full login flow. This is what our monitor loop
        # already does when it detects the login dialog after a
        # session loss — we just invoke it directly.
        log.info("Command server: RECONNECTACCOUNT — re-driving login")
        if _command_server_app is None:
            return "ERR RECONNECTACCOUNT: no app reference"
        ok = attempt_reauth(_command_server_app)
        return "OK RECONNECTACCOUNT" if ok else "ERR RECONNECTACCOUNT: reauth failed"

    if cmd == "ENABLEAPI":
        # We run Gateway with ApiOnly=true in jts.ini, so the API is
        # always enabled once login completes. Nothing to do.
        return "OK ENABLEAPI (already enabled)"

    return f"ERR unknown_command: {cmd}"


def monitor_loop(app):
    """Long-running post-ready monitor.

    Responsibilities:
      - Detect JVM exit and propagate the exit code
      - Heartbeat the API port; if it closes, check for a re-auth dialog
      - If a re-auth dialog is up (daily restart, session expiry), re-drive
        the full login flow

    The heartbeat runs every HEARTBEAT_INTERVAL seconds. Three consecutive
    port-closed heartbeats trigger a re-auth check. This tolerates brief
    network blips (one or two missed heartbeats) without flapping into
    the re-auth path.

    The `app` and `gateway_proc` references are re-read from the module
    globals GATEWAY_PROC / CURRENT_APP each iteration so that a RESTART
    command (Phase 2.4) which replaces the underlying Gateway JVM
    transparently hands the new references to this loop without having
    to be restarted itself.
    """
    global _upstream_dead_jvm
    HEARTBEAT_INTERVAL = 30          # seconds between port probes
    PORT_FAILURE_THRESHOLD = 3       # consecutive failures before we act
    # How many consecutive port-closed heartbeats we'll tolerate when
    # no login dialog is visible before we escalate. This is the
    # "Gateway is wedged in some non-login state with the port closed"
    # case — the controller can't re-drive login because there's no
    # login dialog to drive, so we eventually force-restart the JVM
    # via do_restart_in_place(). This is the escape hatch for the
    # previously-observed dead-end where the loop just kept logging
    # "probably a transient issue, will re-check next heartbeat"
    # forever.
    WEDGED_ESCALATION_THRESHOLD = 6  # = 6 heartbeats × 30s = 3 minutes of wedge

    api_port = api_port_for_mode()
    consecutive_failures = 0
    wedged_failures = 0
    last_heartbeat = time.monotonic()
    last_upstream_check = time.monotonic()
    log.info(f"Monitor: JVM pid={GATEWAY_PROC.pid}, heartbeat API port {api_port} every {HEARTBEAT_INTERVAL}s")

    while True:
        # A halt raised on another thread (a login the command server
        # started) stops this loop acting too; that thread keeps
        # repeating the reason.
        if _current_state == State.HALTED:
            time.sleep(5)
            continue

        # Always consult the live globals — a RESTART command may have
        # replaced them since the last iteration.
        gw = GATEWAY_PROC
        live_app = CURRENT_APP if CURRENT_APP is not None else app

        # During upstream recovery a dead JVM is the watchdog's to relaunch
        # on its schedule — but Gateway's own nightly restart (issue #23)
        # must still be adopted, or the next relaunch would start a second
        # Gateway beside it. Once per exited JVM.
        if (_upstream_recovery and gw is not None and gw.poll() is not None
                and gw is not _upstream_dead_jvm):
            _upstream_dead_jvm = gw
            try:
                _adopt_self_restarted_gateway(
                    "JVM exited during upstream recovery",
                    exit_code=getattr(gw, "returncode", None))
            except Exception as e:
                log.warning(f"Upstream: adoption check raised "
                            f"{type(e).__name__}: {e}")
            continue

        # JVM process check — fast, every iteration.
        if (gw is None or gw.poll() is not None) and not _upstream_recovery:
            rc = gw.returncode if gw is not None else 1
            if isinstance(gw, _AdoptedProcess):
                # Issue #23: not our child, so waitpid can't tell us the
                # exit status. Its next self-restart is still detected
                # via restarter.log, not via the exit code.
                log.error(f"Gateway JVM pid={gw.pid} exited (exit status "
                          "not observable: adopted after Gateway's "
                          "self-restart, not a child of the controller)")
                reason = "adopted Gateway JVM exited"
            elif rc == 0:
                # A scheduled Lock and Exit logoff exits cleanly, and so
                # does Gateway's own restart. Logging that at ERROR is how
                # operators learn to ignore ERROR (field report
                # 2026-09-27). Recovery below is unchanged either way.
                _sched = (os.environ.get("AUTO_LOGOFF_TIME", "").strip()
                          or os.environ.get("AUTO_RESTART_TIME", "").strip())
                log.info(
                    f"Gateway JVM exited cleanly (code 0)"
                    + (f"; a scheduled Lock and Exit is configured "
                       f"({_sched}), so this is expected at that boundary"
                       if _sched else "")
                    + ". Recovering.")
                reason = "JVM exited cleanly (code 0)"
            else:
                log.error(f"Gateway JVM exited with code {rc}")
                reason = f"JVM exited with code {rc}"
            try:
                os.unlink(READY_FILE)
            except FileNotFoundError:
                pass
            # v0.4.7: dual-mode-safe recovery. sys.exit(rc) here would
            # leave this mode's JVM dead while the container stays up
            # on the other mode's PID — same trap v0.4.5/v0.4.6 fixed
            # for the CCP paths. _recover_jvm_or_escalate tries a fast
            # restart first (cheap if IBKR just kicked the session) and
            # falls through to long cool-down if CCP is actually locked.
            # v0.5.10: pass exit_code so the recovery path can apply the
            # IBKR maintenance-window guard when rc==0.
            _recover_jvm_or_escalate(reason, exit_code=rc)
            consecutive_failures = 0
            wedged_failures = 0
            last_heartbeat = time.monotonic()
            continue

        now = time.monotonic()
        if now - last_upstream_check >= _UPSTREAM_POLL_SECONDS:
            last_upstream_check = now
            if _upstream_tick(now):
                _upstream_relaunch()
                consecutive_failures = 0
                wedged_failures = 0
                last_heartbeat = time.monotonic()
                continue

        if not _upstream_recovery:
            boundary = _logoff_backstop_boundary()
            if boundary is not None:
                _run_logoff_backstop(boundary)
                continue

        if _upstream_recovery:
            # The port is expected to be closed while the watchdog waits to
            # relaunch; the port-closed path would escalate to a CCP halt.
            consecutive_failures = 0
            wedged_failures = 0
            last_heartbeat = now
        elif now - last_heartbeat >= HEARTBEAT_INTERVAL:
            last_heartbeat = now
            if is_api_port_open(api_port):
                if consecutive_failures > 0:
                    log.info(f"API port {api_port} recovered after {consecutive_failures} failed heartbeat(s)")
                consecutive_failures = 0
                wedged_failures = 0
            else:
                consecutive_failures += 1
                log.warning(f"API port {api_port} closed "
                            f"(heartbeat failure {consecutive_failures}/{PORT_FAILURE_THRESHOLD})")
                if consecutive_failures >= PORT_FAILURE_THRESHOLD:
                    log.warning("Sustained API port closure — checking for re-auth")
                    result = attempt_reauth(live_app)
                    if result is True:
                        # attempt_reauth returns True both when it re-auth'd
                        # AND when there was nothing to do (no login dialog).
                        # Distinguish: if port is actually open now, we're
                        # recovered; otherwise we're still wedged and should
                        # count this heartbeat as a wedged failure.
                        if is_api_port_open(api_port):
                            consecutive_failures = 0
                            wedged_failures = 0
                            _note_session_login()
                        else:
                            wedged_failures += 1
                            log.warning(f"Wedged failure {wedged_failures}/{WEDGED_ESCALATION_THRESHOLD}: "
                                        "API port still closed, no login dialog")
                            if wedged_failures >= WEDGED_ESCALATION_THRESHOLD:
                                log.error("Gateway appears wedged (sustained "
                                          "port closure, no login dialog "
                                          "appearing). Escalating to "
                                          "do_restart_in_place()")
                                try:
                                    if do_restart_in_place():
                                        log.info("Wedge recovered via in-place restart")
                                        consecutive_failures = 0
                                        wedged_failures = 0
                                    else:
                                        # v0.4.7: don't sys.exit here —
                                        # in dual-mode the container stays
                                        # up on the other PID and this
                                        # mode's JVM is orphaned.
                                        log.error("In-place restart "
                                                  "returned False; "
                                                  "escalating to long "
                                                  "cool-down")
                                        _escalate_to_jvm_restart(
                                            "wedge do_restart_in_place failed")
                                        consecutive_failures = 0
                                        wedged_failures = 0
                                except Exception as e:
                                    log.error(f"In-place restart raised: "
                                              f"{type(e).__name__}: {e}")
                                    # v0.4.7: same reason — escalate
                                    # instead of sys.exit.
                                    _escalate_to_jvm_restart(
                                        f"wedge do_restart_in_place raised "
                                        f"{type(e).__name__}")
                                    consecutive_failures = 0
                                    wedged_failures = 0
                    else:
                        # v0.4.7: reauth failed => dual-mode-safe recovery
                        # (was sys.exit(1), which is a no-op in dual-mode).
                        log.error("Re-auth failed; attempting dual-mode-"
                                  "safe recovery")
                        try:
                            os.unlink(READY_FILE)
                        except FileNotFoundError:
                            pass
                        _recover_jvm_or_escalate("monitor_loop re-auth failed")
                        consecutive_failures = 0
                        wedged_failures = 0

        _sleep_watching_incoming(5)


def attempt_reauth(app):
    """If a new login dialog has appeared during the monitor loop, re-drive
    the full login sequence. Returns True if re-auth completed (or was
    unnecessary), False if it definitively failed."""
    texts, _ = agent_list()
    if "Username" not in texts and "Password" not in texts:
        # API port is closed but no login dialog is on screen. Could be a
        # transient network issue, a shutdown-in-progress, or silent session
        # loss. Nothing to do here — the next heartbeat will re-check.
        log.warning("API port closed but no login dialog visible — "
                    "probably a transient issue, will re-check next heartbeat")
        return True

    log.info("Login dialog detected during monitor loop — re-driving login")
    return _redrive_login(app)


def _redrive_login(app):
    """Re-drive the full login pipeline on a login dialog that is already
    showing. Split out of ``attempt_reauth`` (issue #23) so the
    self-restart adoption path can drive a login on the adopted JVM
    when Gateway did not preserve the session. Returns True if re-auth
    completed (or a CCP backoff was applied and the monitor loop should
    simply retry), False if it definitively failed."""
    # Refresh the app reference (the old one may have gone stale).
    # Scope to our own JVM so dual-mode containers don't cross-drive the
    # other instance's login.
    fresh_app = find_app(APP_NAME_CANDIDATES, timeout=30, match_pid=JVM_PID) or app

    # Clear the readiness file while we're re-authing so consumers know
    # not to use the API port
    try:
        os.unlink(READY_FILE)
    except FileNotFoundError:
        pass

    if not handle_login(fresh_app):
        log.error("Re-auth: login failed")
        return False

    # Check for CCP lockout before burning 120s+ on 2FA/API waits
    if _detect_ccp_lockout(timeout=25):
        _apply_ccp_backoff()
        log.info("Re-auth: CCP lockout detected — backing off. "
                 "Monitor loop will retry on next heartbeat cycle.")
        return True  # let the monitor loop re-check after the backoff

    # Gate: same reasoning as the main() path — don't reset on the
    # stuck-connecting case, only on genuine CCP-gate progress.
    if not _detect_login_stuck_connecting():
        _reset_ccp_backoff()

    if not handle_post_login_dialogs(fresh_app):
        log.error("Re-auth: post-login dialogs failed")
        return False
    if not handle_2fa(fresh_app):
        log.error("Re-auth: 2FA failed")
        return False
    dismiss_post_login_disclaimers(timeout=30)
    if not wait_for_api_port(timeout=120):
        log.error("Re-auth: API port never came back up")
        return False

    signal_ready()
    log.info("Re-auth complete, resuming monitor")
    return True


if __name__ == "__main__":
    main()