#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""AlphaBot V2 native broker execution candidate, 1.0.0-rc3.

Real Alpaca REST execution + authenticated data/order WebSockets; Kraken signed
REST + authenticated WebSocket v2 execution. One dedicated USD long-only spot
account per project. Independent OS thread owns the 1 Hz risk loop. No backtest
or demo execution path. `check` never submits orders. `run --live` does.
"""
import argparse
import asyncio
import base64
import contextlib
from datetime import datetime, timezone, timedelta
from decimal import Decimal, ROUND_DOWN, getcontext
import fcntl
import getpass
import platform
import resource
import ssl
import sys
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import signal
import sqlite3
import threading
import time
import urllib.parse
import uuid

import aiohttp
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from yarl import URL
import audit_ledger
import audit_viewer
from audit_dashboard import Observer
from host_strategy import capital_floor, decide, decimal, risk_multiplier, target_deviation

VERSION = '1.0.0-rc3'
D = Decimal
getcontext().prec = 50
ZERO = D('0')
TERMINAL = {'filled', 'canceled', 'cancelled', 'expired', 'rejected', 'closed'}


class AuthenticationFailed(Exception):
    def __init__(self):
        super().__init__('AUTHENTICATION_FAILED')


class LocalVault:
    """Explicit v1 local vault: scrypt + AES-GCM, RFC 6238, durable replay guard.

    This defines a new documented adapter; it does not guess an unseen vault's
    schema. Import an existing Authenticator Base32 seed through enroll-vault.
    """
    AAD = b'AlphaBot local authentication vault v1'

    def __init__(self, path):
        self.path = Path(path).expanduser()
        self.state_path = self.path.with_name(self.path.name + '.state.sqlite3')

    @staticmethod
    def key(password, salt):
        return hashlib.scrypt(password.encode('utf-8'), salt=salt, n=32768, r=8, p=1,
                              dklen=32, maxmem=128 * 1024 * 1024)

    @staticmethod
    def seed(text):
        normalized = text.strip().replace(' ', '').upper().rstrip('=')
        value = base64.b32decode(normalized + '=' * ((-len(normalized)) % 8), casefold=False)
        if len(value) < 20:
            raise ValueError('Authenticator seed must contain at least 160 bits')
        return value

    @classmethod
    def token(cls, seed, counter):
        digest = hmac.new(cls.seed(seed), int(counter).to_bytes(8, 'big'), hashlib.sha1).digest()
        offset = digest[-1] & 15
        value = int.from_bytes(digest[offset:offset+4], 'big') & 0x7fffffff
        return f'{value % 1000000:06d}'

    def create(self, password, seed):
        self.seed(seed)
        if not password:
            raise ValueError('Empty app password is not allowed')
        salt, nonce = os.urandom(16), os.urandom(12)
        plain = json.dumps({'version': 1, 'totp_secret': seed, 'algorithm': 'SHA1',
                            'digits': 6, 'period': 30}).encode()
        encrypted = AESGCM(self.key(password, salt)).encrypt(nonce, plain, self.AAD)
        value = {'version': 1, 'kdf': 'scrypt-32768-8-1',
                 'salt': base64.b64encode(salt).decode(),
                 'nonce': base64.b64encode(nonce).decode(),
                 'ciphertext': base64.b64encode(encrypted).decode()}
        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
        descriptor = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(descriptor, 'w') as stream:
            json.dump(value, stream)
            stream.flush()
            os.fsync(stream.fileno())

    def verify(self, password, token, at=None):
        connection = None
        try:
            if not isinstance(token, str) or not re.fullmatch(r'[0-9]{6}', token):
                raise ValueError
            if self.path.is_symlink() or self.path.stat().st_mode & 0o077:
                raise ValueError
            if self.path.stat().st_size > 16384:
                raise ValueError
            envelope = json.loads(self.path.read_text())
            if envelope.get('version') != 1 or envelope.get('kdf') != 'scrypt-32768-8-1':
                raise ValueError
            salt, nonce, ciphertext = (base64.b64decode(envelope[k], validate=True)
                                        for k in ('salt', 'nonce', 'ciphertext'))
            if len(salt) != 16 or len(nonce) != 12:
                raise ValueError
            plain = AESGCM(self.key(password, salt)).decrypt(nonce, ciphertext, self.AAD)
            account = json.loads(plain)
            if (account.get('version') != 1 or account.get('algorithm') != 'SHA1' or
                    account.get('digits') != 6 or account.get('period') != 30):
                raise ValueError
            counter = int(time.time() if at is None else at) // 30
            matches = [n for n in (counter - 1, counter, counter + 1) if n >= 0 and
                       hmac.compare_digest(self.token(account['totp_secret'], n), token)]
            if not matches:
                raise ValueError
            accepted = max(matches)
            if self.state_path.is_symlink():
                raise ValueError
            descriptor = os.open(self.state_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
            os.close(descriptor)
            os.chmod(self.state_path, 0o600)
            connection = sqlite3.connect(self.state_path, timeout=5)
            connection.execute('PRAGMA synchronous=FULL')
            connection.execute('CREATE TABLE IF NOT EXISTS replay(id INTEGER PRIMARY KEY CHECK(id=1),counter INTEGER NOT NULL)')
            connection.execute('BEGIN IMMEDIATE')
            prior = connection.execute('SELECT counter FROM replay WHERE id=1').fetchone()
            if prior and accepted <= prior[0]:
                raise ValueError
            connection.execute('INSERT OR REPLACE INTO replay VALUES(1,?)', (accepted,))
            connection.commit()
        except Exception:
            raise AuthenticationFailed() from None
        finally:
            if connection:
                connection.close()
            password = token = None


class AdministratorGate:
    PROMPT = 'Enter App-Level Secret Password: '
    MFA_PROMPT = 'Enter 6-Digit Google Authenticator MFA Token: '
    TTL = 3600

    def __init__(self, settings, environment, on_failure=None):
        self.settings, self.on_failure = settings, on_failure
        supplied = environment.get('LIVE_AUTH_GATEWAY_URL', '')
        if supplied and '://' not in supplied:
            supplied = 'https://' + supplied
        try:
            parsed = urllib.parse.urlsplit(supplied)
            if (parsed.scheme != 'https' or not parsed.hostname or parsed.username or
                parsed.password or parsed.query or parsed.fragment or
                parsed.path not in ('', '/', '/api/v1/auth/gate')):
                raise ValueError
            if settings.get('gateway_hostname') and parsed.hostname != settings['gateway_hostname']:
                raise ValueError
            if settings.get('gateway_port') and (parsed.port or 443) != settings['gateway_port']:
                raise ValueError
            parsed.port
        except ValueError:
            raise AuthenticationFailed() from None
        self.hostname = parsed.hostname
        self.base = urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, '', '', ''))
        self.headers = {}
        cf_id = environment.get('CF_ACCESS_CLIENT_ID')
        cf_secret = environment.get('CF_ACCESS_CLIENT_SECRET')
        if bool(cf_id) != bool(cf_secret):
            raise AuthenticationFailed()
        if cf_id:
            self.headers = {'CF-Access-Client-Id': cf_id, 'CF-Access-Client-Secret': cf_secret}
        self.ca_file = environment.get('LIVE_AUTH_GATEWAY_CA_FILE')
        self.vault = LocalVault(settings.get('vault_path', 'auth.vault.json'))
        self.revoked, self.closed = threading.Event(), threading.Event()
        self.revoked.set()
        self.deadline, self.generation = 0, 0
        self.session = self.bound_account = self.monitor = None
        self.mutex = threading.RLock()

    @staticmethod
    def prompt_factors():
        if not sys.stdin.isatty() or threading.current_thread() is not threading.main_thread():
            raise AuthenticationFailed()
        resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
        password = getpass.getpass(AdministratorGate.PROMPT)
        token = getpass.getpass(AdministratorGate.MFA_PROMPT)
        return password, token

    def _prompt(self):
        return self.prompt_factors()

    async def open(self, factors=None):
        options = {}
        if self.ca_file:
            options['connector'] = aiohttp.TCPConnector(ssl=ssl.create_default_context(cafile=self.ca_file))
        self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15),
            trust_env=False, cookie_jar=aiohttp.CookieJar(), **options)
        try:
            await self.authenticate(factors)
        except BaseException:
            await self.close()
            raise
        self.monitor = threading.Thread(target=self._watch, name='administrator-expiry', daemon=True)
        self.monitor.start()
        return self

    def _watch(self):
        while not self.closed.wait(0.1):
            with self.mutex:
                if time.monotonic() >= self.deadline:
                    self.revoked.set()

    def require(self):
        with self.mutex:
            if self.closed.is_set() or self.revoked.is_set() or time.monotonic() >= self.deadline:
                self.revoked.set()
                raise AuthenticationFailed()

    @property
    def expires_in(self):
        with self.mutex:
            return max(0, int(self.deadline - time.monotonic()))

    def _cookie_valid(self):
        names = set(self.session.cookie_jar.filter_cookies(URL(self.base)))
        return any(c.key in names and c['secure'] and c['httponly'] and c['path'] == '/' and
                   c['domain'].lstrip('.') == self.hostname for c in self.session.cookie_jar)

    async def _post(self, path, body, authenticated=True):
        if authenticated:
            self.require()
            if not self._cookie_valid():
                raise AuthenticationFailed()
        async with self.session.post(self.base + path, json=body, headers=self.headers,
                                     allow_redirects=False) as response:
            if response.status != 200:
                raise AuthenticationFailed()
            raw = await response.content.read(65537)
            if len(raw) > 65536:
                raise AuthenticationFailed()
            result = json.loads(raw)
            if not isinstance(result, dict):
                raise AuthenticationFailed()
            return result

    async def authenticate(self, factors=None):
        self.revoked.set()
        self.session.cookie_jar.clear()
        password = token = challenge = None
        try:
            password, token = self._prompt() if factors is None else factors
            self.vault.verify(password, token)
            challenge = hashlib.sha256(password.encode('utf-8')).hexdigest()
            password = token = factors = None
            started = time.monotonic()
            result = await self._post('/api/v1/auth/gate', {'gate_challenge': challenge}, authenticated=False)
            if (result.get('status') != 'AUTHENTICATED' or
                type(result.get('session_token_expires_sec')) is not int or
                result['session_token_expires_sec'] != self.TTL or not self._cookie_valid()):
                raise AuthenticationFailed()
            with self.mutex:
                self.deadline = started + self.TTL
                if self.deadline <= time.monotonic():
                    raise AuthenticationFailed()
                self.generation += 1
                self.revoked.clear()
        except BaseException:
            self.revoked.set()
            self.session.cookie_jar.clear()
            if self.on_failure:
                self.on_failure('AUTHENTICATION_FAILED')
            raise AuthenticationFailed() from None
        finally:
            password = token = challenge = factors = None

    async def bootstrap(self, broker, account_id, key, secret):
        try:
            if not account_id or not key or not secret:
                raise AuthenticationFailed()
            result = await self._post('/v1/broker-session', {'broker': broker, 'account_id': account_id,
                'api_key': key, 'secret_key': secret})
            if result.get('bound') is not True or result.get('account_id') != account_id:
                raise AuthenticationFailed()
            self.bound_account = account_id
        except BaseException:
            self.revoked.set()
            if self.on_failure:
                self.on_failure('AUTHENTICATION_FAILED')
            raise AuthenticationFailed() from None

    async def renew_if_due(self):
        if self.revoked.is_set() or time.monotonic() >= self.deadline:
            await self.authenticate()
        self.require()

    async def close(self):
        self.closed.set()
        self.revoked.set()
        if self.session:
            self.session.cookie_jar.clear()
            await self.session.close()
        if self.monitor:
            self.monitor.join(timeout=0.3)
        self.headers.clear()


class Fault(Exception):
    """A message safe to show: never include upstream bodies or credentials."""


def now():
    return datetime.now(timezone.utc).isoformat()


def timestamp(value):
    if len(value) == 10:
        value += 'T00:00:00+00:00'
    result = datetime.fromisoformat(value.replace('Z', '+00:00'))
    if result.tzinfo is None:
        raise Fault('timezone_required')
    return result.astimezone(timezone.utc)


def fixed(value):
    if isinstance(value, D):
        if not value.is_finite():
            raise ValueError('nonfinite_decimal')
        return format(value, 'f')
    if isinstance(value, dict):
        return {str(k): fixed(v) for k, v in value.items()}
    if isinstance(value, (tuple, list)):
        return [fixed(v) for v in value]
    if isinstance(value, float):
        raise ValueError('binary_float_in_accounting')
    return value


def encode(value):
    return json.dumps(fixed(value), sort_keys=True, separators=(',', ':'), allow_nan=False)


def wire(value):
    """Kraken requires numeric JSON tokens, without a binary-float conversion."""
    if isinstance(value, D):
        return format(decimal(value), 'f')
    if isinstance(value, dict):
        return '{' + ','.join(json.dumps(k) + ':' + wire(v) for k, v in value.items()) + '}'
    if isinstance(value, list):
        return '[' + ','.join(map(wire, value)) + ']'
    return json.dumps(value, allow_nan=False)


def decode(value):
    def invalid(_):
        raise Fault('nonfinite_json')
    return json.loads(value, parse_float=D, parse_constant=invalid)


def log(event, **data):
    print(encode({'at': now(), 'event': event, **data}), flush=True)


def configuration(root):
    cfg = decode((root / 'config.json').read_text())
    if cfg.get('broker') not in ('alpaca', 'kraken'):
        raise Fault('broker_must_be_alpaca_or_kraken')
    symbol = cfg.get('symbol', '')
    if cfg['broker'] == 'alpaca' and not re.fullmatch(r'[A-Z][A-Z0-9.\-]{0,14}', symbol):
        raise Fault('invalid_equity_symbol')
    if cfg['broker'] == 'kraken' and not re.fullmatch(r'[A-Z0-9]+/USD', symbol):
        raise Fault('kraken_requires_BASE_USD')
    defaults = {'starting_capital': '500', 'lesson_risk_decay': '0.95',
        'execution_target_multiplier': '1', 'fee_reserve_rate': '0.005',
        'slippage_reserve_rate': '0.01', 'max_order_usd': '100',
        'market_data_max_age_seconds': 120, 'alpaca_feed': 'iex',
        'dedicated_account': True}
    cfg = {**defaults, **cfg}
    for key in ('starting_capital', 'max_order_usd', 'execution_target_multiplier'):
        if decimal(cfg[key]) <= 0:
            raise Fault('positive_configuration_required_' + key)
    for key in ('fee_reserve_rate', 'slippage_reserve_rate'):
        if not 0 <= decimal(cfg[key]) < D('0.25'):
            raise Fault('invalid_reserve_' + key)
    if cfg['dedicated_account'] is not True:
        raise Fault('dedicated_account_required')
    if cfg['alpaca_feed'] not in ('iex', 'sip'):
        raise Fault('invalid_alpaca_feed')
    if type(cfg['market_data_max_age_seconds']) is not int or not 60 <= cfg['market_data_max_age_seconds'] <= 300:
        raise Fault('market_data_age_must_be_60_to_300_seconds')
    risk_multiplier(0, cfg['lesson_risk_decay'])
    return cfg


class Journal:
    """FULL-sync WAL transactions; shared across risk/strategy via an RLock.

    Observations are an internal management ledger, not a GAAP/tax book. The
    append-only hash chain detects edits, not a privileged attack rewriting it all.
    Pending intents are committed BEFORE network submission; never blindly replay.
    """
    def __init__(self, root, cfg, key_fingerprint, defer_binding_review=False):
        self.mutex = threading.RLock()
        self.halted = threading.Event()
        self.kraken_gate = threading.Lock()
        self.gate = None
        self.observer = None
        self.order_gate = threading.Lock()
        self.binding_mismatch = False
        self.proposed_binding = None
        self.root, self.cfg = root, cfg
        self.lock_file = (root / 'native-live.lock').open('a')
        try:
            fcntl.flock(self.lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            self.lock_file.close()
            raise Fault('project_already_running') from None
        try:
            self.db = sqlite3.connect(root / 'native-live.sqlite3', check_same_thread=False, timeout=5)
            self.db.executescript('''PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
            CREATE TABLE IF NOT EXISTS state(k TEXT PRIMARY KEY,v TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS events(id INTEGER PRIMARY KEY,at TEXT NOT NULL,
              kind TEXT NOT NULL,body TEXT NOT NULL,parent TEXT NOT NULL,hash TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS intents(client_id TEXT PRIMARY KEY,body TEXT NOT NULL,
              status TEXT NOT NULL,broker_id TEXT);
            CREATE TABLE IF NOT EXISTS receipts(broker_id TEXT PRIMARY KEY,body TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS lessons(id INTEGER PRIMARY KEY,event_id INTEGER,
              body TEXT NOT NULL);''')
            os.chmod(root / 'native-live.sqlite3', 0o600)
            audit_ledger.install(self.db)
            self.verify()
            binding = hashlib.sha256(encode({'config': cfg, 'key_fingerprint': key_fingerprint}).encode()).hexdigest()
            with self.mutex, self.db:
                old = self.get('binding')
                self.proposed_binding = binding
                self.binding_mismatch = bool(old and old != binding)
                if self.binding_mismatch and not defer_binding_review:
                    raise Fault('durable_account_or_config_binding_mismatch')
                if not self.binding_mismatch:
                    self.set('binding', binding)
                if self.get('halt'):
                    self.halted.set()
                if self.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0]:
                    self.halt('restart_with_unresolved_order_intent')
        except BaseException:
            with contextlib.suppress(Exception):
                self.db.close()
            self.lock_file.close()
            raise

    def get(self, key, default=None):
        with self.mutex:
            row = self.db.execute('SELECT v FROM state WHERE k=?', (key,)).fetchone()
            return decode(row[0]) if row else default

    def set(self, key, value):
        self.db.execute('INSERT OR REPLACE INTO state VALUES(?,?)', (key, encode(value)))

    def _append(self, kind, body):
        at = now()
        parent = self.db.execute('SELECT hash FROM events ORDER BY id DESC LIMIT 1').fetchone()
        parent = parent[0] if parent else '0' * 64
        body = encode(body)
        digest = hashlib.sha256(encode([at, kind, body, parent]).encode()).hexdigest()
        cursor = self.db.execute('INSERT INTO events(at,kind,body,parent,hash) VALUES(?,?,?,?,?)',
                                 (at, kind, body, parent, digest))
        event_id = cursor.lastrowid
        for seq, line in enumerate(decode(body).get('journal', [])):
            self.db.execute('INSERT INTO journal_lines VALUES(?,?,?,?,?,?)',
                (event_id, seq, line['account'], line['debit'], line['credit'], line['memo']))
        return event_id

    def append(self, kind, body):
        with self.mutex, self.db:
            return self._append(kind, body)

    def verify(self):
        with self.mutex:
            try:
                return audit_viewer.verify(self.db)
            except ValueError as exc:
                raise Fault(str(exc)) from None

    def lesson_count(self):
        with self.mutex:
            return self.db.execute('SELECT count(*) FROM lessons').fetchone()[0]

    def _lesson(self, path, details):
        count = self.lesson_count()
        prior = self.db.execute('SELECT body FROM lessons ORDER BY id DESC LIMIT 1').fetchone()
        parent = decode(prior[0]).get('lesson_hash') if prior else None
        body = {'parent_lesson_hash': parent, 'risk_formula': 'decay**cumulative_count',
                'failure_path': path, 'details': details, 'cumulative_count': count + 1,
                'multiplier_before': risk_multiplier(count, self.cfg['lesson_risk_decay']),
                'multiplier_after': risk_multiplier(count + 1, self.cfg['lesson_risk_decay'])}
        body['lesson_hash'] = hashlib.sha256(encode(body).encode()).hexdigest()
        event_id = self._append('LESSON', body)
        self.db.execute('INSERT INTO lessons(event_id,body) VALUES(?,?)', (event_id, encode(body)))

    def halt(self, reason):
        # Admission stops even if a database commit fails; I/O failure must not allow a buy.
        self.halted.set()
        with self.mutex, self.db:
            if not self.get('halt'):
                self.set('halt', {'at': now(), 'reason': reason})
                body = {'reason': reason, 'journal': [], 'reference_price_available': False}
                book = self.get('book')
                if book is not None:
                    body['accounting'] = {'debit_total': '0', 'credit_total': '0', 'book_after': book}
                    body['last_snapshot'] = self.get('last_snapshot')
                self._append('HALT', body)
        log('HALT', reason=reason)

    def intent(self, side, quantity, reference, month='', emergency=False):
        with self.mutex, self.db:
            if (self.halted.is_set() or self.get('accounting_unresolved') or self.binding_mismatch) and not emergency:
                raise Fault('durable_halt_blocks_entry')
            cid = str(uuid.uuid4())
            body = {'side': side, 'quantity': quantity, 'reference': reference,
                    'month': month, 'symbol': self.cfg['symbol'], 'emergency': emergency}
            self.db.execute('INSERT INTO intents VALUES(?,?,?,NULL)', (cid, encode(body), 'pending'))
            self._append('ORDER_INTENT', {'client_id': cid, **body})
            self.set('execution_generation', int(self.get('execution_generation', 0)) + 1)
            if side == 'buy':
                self.set('last_entry_month', month)  # A partial/ambiguous entry consumes its month.
            return cid

    def receipt(self, cid, result):
        """One terminal aggregate receipt per broker order; no partial double counting."""
        with self.mutex, self.db:
            row = self.db.execute('SELECT body FROM intents WHERE client_id=?', (cid,)).fetchone()
            if not row:
                raise Fault('receipt_without_intent')
            intent = decode(row[0])
            bid = result['broker_id']
            prior = self.db.execute('SELECT body FROM receipts WHERE broker_id=?', (bid,)).fetchone()
            if prior:
                if decode(prior[0]) != fixed(result):
                    raise Fault('conflicting_terminal_receipt')
                return
            self.db.execute('INSERT INTO receipts VALUES(?,?)', (bid, encode(result)))
            self.db.execute('UPDATE intents SET status=?,broker_id=? WHERE client_id=?',
                            (result['status'], bid, cid))
            qty, price = decimal(result['quantity']), decimal(result['price'])
            prior_quantity = decimal(self.get('quantity', '0'))
            prior_cost = decimal(self.get('asset_cost', '0'))
            released_cost = min(qty, prior_quantity) * prior_cost / prior_quantity if prior_quantity else ZERO
            if qty > 0:
                held, cost = decimal(self.get('quantity', '0')), decimal(self.get('asset_cost', '0'))
                fee = result.get('fee')
                fee_cost = decimal(fee) if fee is not None else qty * price * decimal(self.cfg['fee_reserve_rate'])
                cash_low = decimal(self.get('cash_lower', self.cfg['starting_capital']))
                cash_high = decimal(self.get('cash_upper', self.cfg['starting_capital']))
                gross = qty * price
                minimum_fee = decimal(fee) if fee is not None else ZERO
                if intent['side'] == 'buy':
                    cash_low, cash_high = cash_low - gross - fee_cost, cash_high - gross - minimum_fee
                else:
                    cash_low, cash_high = cash_low + gross - fee_cost, cash_high + gross - minimum_fee
                self.set('cash_lower', cash_low)
                self.set('cash_upper', cash_high)
                if intent['side'] == 'buy':
                    held, cost = held + qty, cost + qty * price + fee_cost
                else:
                    if qty > held + D('0.000000001'):
                        self.halt('sell_exceeds_accounted_holdings')
                    remaining = max(ZERO, held - qty)
                    cost = cost * remaining / held if held else ZERO
                    held = remaining
                self.set('quantity', held)
                self.set('asset_cost', cost)
                target, deviation = target_deviation(decimal(intent['reference']), price,
                    self.cfg['execution_target_multiplier'])
                if deviation >= D('0.0001'):
                    self._lesson('order.fill.price_vs_target', {'client_id': cid, 'broker_id': bid,
                        'target': target, 'fill': price, 'deviation_pct': deviation,
                        'threshold_pct': D('0.0001'), 'fee_basis': result['fee_basis']})
            accounting_body = audit_ledger.execution(self, {'client_id': cid, **result}, intent, result, released_cost, prior_quantity)
            self._append('EXECUTION_RECEIPT', accounting_body)
            self.set('execution_generation', int(self.get('execution_generation', 0)) + 1)

    def observe(self, kind, snapshot, **extra):
        with self.mutex, self.db:
            equity = decimal(snapshot['equity'])
            peak = max(equity, decimal(self.get('peak', self.cfg['starting_capital'])))
            floor, ratchet = capital_floor(self.get('protected_starting_capital', self.cfg['starting_capital']),
                self.get('asset_cost', '0'), peak, self.get('ratchet', '250'))
            self.set('peak', peak)
            self.set('ratchet', ratchet)
            reserve = decimal(snapshot['gross_asset_value']) * (
                decimal(self.cfg['fee_reserve_rate']) + decimal(self.cfg['slippage_reserve_rate']))
            net = equity - reserve
            count = self.lesson_count()
            body = {'snapshot': snapshot, 'peak': peak, 'milestone_floor': ratchet,
                'floor': floor, 'net_liquidation_estimate': net,
                'asset_cost_including_fee_or_reserve': self.get('asset_cost', '0'),
                'lesson_count': count, 'risk_multiplier': risk_multiplier(count, self.cfg['lesson_risk_decay']),
                'halted': self.halted.is_set(), **extra}
            body = audit_ledger.observe(self, body, snapshot)
            body['lesson_count'] = self.lesson_count()
            body['risk_multiplier'] = risk_multiplier(self.lesson_count(), self.cfg['lesson_risk_decay'])
            self._append(kind, body)
            self.set('last_snapshot', snapshot)
            if self.get('accounting_unresolved') and not self.get('halt'):
                self.halted.set()
                reason = 'unclassified_accounting_movement'
                self.set('halt', {'at': now(), 'reason': reason})
                self._append('HALT', {'reason': reason, 'journal': [],
                    'accounting': {'debit_total':'0','credit_total':'0','book_after':self.get('book')}})
            return net, floor

    def close(self):
        with self.mutex:
            self.db.close()
            fcntl.flock(self.lock_file, fcntl.LOCK_UN)
            self.lock_file.close()


class Broker:
    def __init__(self, cfg, key, secret, journal):
        self.cfg, self.key, self.secret, self.journal = cfg, key, secret, journal
        self.symbol = cfg['symbol']
        self.session = None
        self.stream_ready = asyncio.Event()

    async def __aenter__(self):
        self.require_auth()
        self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=8),
            trust_env=False, raise_for_status=False)
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    def require_auth(self, bound=False):
        if self.journal.gate is None:
            raise AuthenticationFailed()
        self.journal.gate.require()
        if bound and not self.journal.gate.bound_account:
            raise AuthenticationFailed()

    def protected_ws(self, *args, **kwargs):
        self.require_auth()
        return self.session.ws_connect(*args, **kwargs)

    async def http(self, method, url, **kwargs):
        self.require_auth()
        try:
            async with self.session.request(method, url, allow_redirects=False, **kwargs) as response:
                if response.status >= 300:
                    raise Fault('broker_http_' + str(response.status))
                raw = await response.content.read(4 * 1024 * 1024 + 1)
                if len(raw) > 4 * 1024 * 1024:
                    raise Fault('broker_response_too_large')
                return decode(raw) if raw else {}
        except (aiohttp.ClientError, asyncio.TimeoutError):
            raise Fault('broker_transport_failure') from None

    async def frames(self, socket):
        async for message in socket:
            if message.type in (aiohttp.WSMsgType.TEXT, aiohttp.WSMsgType.BINARY):
                self.require_auth()
                yield decode(message.data)
            elif message.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
                break
        raise Fault('broker_socket_closed')

    async def terminal(self, broker_id, timeout=15):
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            result = await self.order_info(broker_id)
            if result['status'] in TERMINAL:
                return result
            await asyncio.sleep(DURATION_POLL)
        raise Fault('order_reconciliation_timeout')


DURATION_POLL = 0.5  # Scheduling intervals only; no money passes through float.


class Alpaca(Broker):
    REST = 'https://api.alpaca.markets'
    DATA = 'wss://stream.data.alpaca.markets/v2/'
    ORDERS = 'wss://api.alpaca.markets/stream'

    def headers(self):
        return {'APCA-API-KEY-ID': self.key, 'APCA-API-SECRET-KEY': self.secret}

    async def rest(self, method, path, **kwargs):
        return await self.http(method, self.REST + path, headers=self.headers(), **kwargs)

    async def prepare(self):
        account = await self.rest('GET', '/v2/account')
        if account.get('currency') != 'USD' or account.get('status') != 'ACTIVE':
            raise Fault('active_USD_account_required')
        if account.get('trading_blocked') or decimal(account.get('short_market_value', '0')) != 0:
            raise Fault('blocked_or_short_account')
        if decimal(account.get('multiplier', '1')) != 1:
            raise Fault('disable_margin_use_cash_account')
        old = self.journal.get('broker_account_id')
        if old and old != account['id']:
            raise Fault('broker_account_identity_changed')
        with self.journal.mutex, self.journal.db:
            self.journal.set('broker_account_id', account['id'])
        asset = await self.rest('GET', '/v2/assets/' + urllib.parse.quote(self.symbol, safe=''))
        if asset.get('tradable') is not True or asset.get('fractionable') is not True:
            raise Fault('tradable_fractionable_equity_required')

    async def snapshot(self):
        account, positions = await asyncio.gather(self.rest('GET', '/v2/account'),
                                                 self.rest('GET', '/v2/positions'))
        if account['id'] != self.journal.get('broker_account_id'):
            raise Fault('account_identity_mismatch')
        if account.get('currency') != 'USD' or decimal(account.get('multiplier', '1')) != 1:
            raise Fault('account_scope_changed')
        foreign = [p['symbol'] for p in positions if p['symbol'] != self.symbol or p['side'] != 'long']
        if foreign:
            raise Fault('unsupported_or_external_position')
        qty = sum((decimal(p['qty']) for p in positions), ZERO)
        gross = sum((decimal(p['market_value']) for p in positions), ZERO)
        return {'at': now(), 'cash': decimal(account['cash']), 'quantity': qty,
            'equity': decimal(account['equity']), 'gross_asset_value': gross,
            'source': 'alpaca_REST_account_and_positions',
            'positions': [{'symbol':self.symbol,'quantity':qty,'price':gross/qty,
                'cost':decimal(self.journal.get('asset_cost','0')),
                'cost_basis':'fee_inclusive_actual_or_disclosed_reserve'}] if qty>0 else [],
            'equity_reconciliation_difference':decimal(account['equity'])-decimal(account['cash'])-gross}

    async def orders(self):
        return await self.rest('GET', '/v2/orders?status=open&limit=500')

    async def buy_allowed(self):
        clock = await self.rest('GET', '/v2/clock')
        if clock.get('is_open') is not True:
            raise Fault('equity_market_closed')

    async def submit(self, cid, side, qty):
        self.require_auth(bound=True)
        row = await self.rest('POST', '/v2/orders', json={'symbol': self.symbol,
            'qty': format(qty, 'f'), 'side': side, 'type': 'market',
            'time_in_force': 'day', 'client_order_id': cid})
        if row.get('client_order_id') != cid:
            raise Fault('order_identity_mismatch')
        return row['id']

    async def find_order(self, cid):
        row = await self.rest('GET', '/v2/orders:by_client_order_id', params={'client_order_id': cid})
        if row.get('client_order_id') != cid:
            raise Fault('recovery_order_identity_mismatch')
        return row['id']

    async def order_info(self, bid):
        row = await self.rest('GET', '/v2/orders/' + urllib.parse.quote(bid, safe=''))
        qty = decimal(row.get('filled_qty', '0'))
        return {'broker_id': row['id'], 'status': row['status'], 'quantity': qty,
            'price': decimal(row.get('filled_avg_price') or '0'), 'fee': None,
            'fee_basis': 'fee_unknown_conservative_configured_reserve', 'at': now()}

    async def cancel_all(self):
        await self.rest('DELETE', '/v2/orders')

    async def bars(self):
        async with self.protected_ws(self.DATA + self.cfg['alpaca_feed'], heartbeat=20,
                max_msg_size=2 * 1024 * 1024) as ws:
            await ws.send_json({'action': 'auth', 'key': self.key, 'secret': self.secret})
            subscribed = False
            async for packet in self.frames(ws):
                for row in packet:
                    if row.get('T') == 'error':
                        raise Fault('alpaca_market_stream_rejected')
                    if row.get('T') == 'success' and row.get('msg') == 'authenticated':
                        await ws.send_json({'action': 'subscribe', 'bars': [self.symbol]})
                    if row.get('T') == 'subscription':
                        if self.symbol not in row.get('bars', []):
                            raise Fault('alpaca_bar_subscription_missing')
                        subscribed = True
                    if row.get('T') == 'b' and row.get('S') == self.symbol and subscribed:
                        yield {'at': row['t'], 'price': decimal(row['c']), 'source': 'alpaca_bar'}

    async def account_stream(self):
        async with self.protected_ws(self.ORDERS, heartbeat=20) as ws:
            await ws.send_json({'action': 'auth', 'key': self.key, 'secret': self.secret})
            async for row in self.frames(ws):
                if row.get('stream') == 'authorization':
                    if row.get('data', {}).get('status') != 'authorized':
                        raise Fault('alpaca_order_stream_auth_failed')
                    await ws.send_json({'action': 'listen', 'data': {'streams': ['trade_updates']}})
                elif row.get('stream') == 'listening':
                    if 'trade_updates' not in row.get('data', {}).get('streams', []):
                        raise Fault('alpaca_order_subscription_missing')
                    self.stream_ready.set()
                elif row.get('stream') == 'trade_updates':
                    data = row['data']
                    # Persist selected nonsecret fields, never arbitrary account payloads.
                    self.journal.append('ORDER_STREAM', {'event': data['event'],
                        'broker_id': data['order']['id'], 'status': data['order']['status'],
                        'filled_qty': decimal(data['order'].get('filled_qty', '0'))})


class Kraken(Broker):
    REST = 'https://api.kraken.com'
    PUBLIC = 'wss://ws.kraken.com/v2'
    PRIVATE = 'wss://ws-auth.kraken.com/v2'

    def __init__(self, *args):
        super().__init__(*args)
        self.balance = {}
        self.balance_ready = asyncio.Event()
        self.last_private_message = 0
        self.sequence = None
        self.ws = None
        self.ws_pending = {}
        self.req = 0
        self.pair = None
        self.base_code = None
        self.base_name = self.symbol.split('/')[0]
        self.token = None

    def signed(self, path, data):
        # Nonce is durable and serialized across the two broker threads/process restarts.
        with self.journal.mutex, self.journal.db:
            nonce = max(time.time_ns() // 1000, int(self.journal.get('nonce', 0)) + 1)
            self.journal.set('nonce', nonce)
        body = urllib.parse.urlencode({'nonce': str(nonce), **data})
        digest = hashlib.sha256((str(nonce) + body).encode()).digest()
        try:
            secret = base64.b64decode(self.secret, validate=True)
        except ValueError:
            raise Fault('kraken_secret_must_be_base64') from None
        signature = base64.b64encode(hmac.new(secret, path.encode() + digest, hashlib.sha512).digest()).decode()
        return body, {'API-Key': self.key, 'API-Sign': signature,
                      'Content-Type': 'application/x-www-form-urlencoded'}

    async def rest(self, endpoint, data=None, private=True):
        path = '/0/' + ('private/' if private else 'public/') + endpoint
        if private:
            # Separate per-process nonce serialization includes request completion so a
            # later nonce cannot arrive before an earlier one on another thread.
            async with thread_gate(self.journal.kraken_gate):
                body, headers = self.signed(path, data or {})
                row = await self.http('POST', self.REST + path, data=body, headers=headers)
        else:
            row = await self.http('GET', self.REST + path, params=data or {})
        if row.get('error'):
            raise Fault('kraken_request_rejected')
        return row['result']

    async def prepare(self):
        pairs = await self.rest('AssetPairs', private=False)
        wanted = ('XBT' if self.base_name == 'BTC' else self.base_name) + '/USD'
        matches = [(code, p) for code, p in pairs.items() if p.get('wsname') == wanted]
        if len(matches) != 1:
            raise Fault('kraken_pair_not_supported')
        self.pair, info = matches[0]
        if info.get('status') != 'online':
            raise Fault('kraken_pair_not_online')
        self.base_code = info['base']
        self.qty_step = D('1').scaleb(-int(info['lot_decimals']))
        self.minimum = decimal(info['ordermin'])
        self.cost_minimum = decimal(info.get('costmin', '0'))
        token = await self.rest('GetWebSocketsToken')
        self.token = token['token']
        if await self.rest('OpenPositions'):
            raise Fault('kraken_margin_positions_unsupported')

    def validate_balances(self, balances):
        allowed = {'ZUSD', 'USD', self.base_code, self.base_name,
                   'XBT' if self.base_name == 'BTC' else self.base_name}
        for asset, amount in balances.items():
            if decimal(amount) < 0 or (decimal(amount) != 0 and asset not in allowed):
                raise Fault('kraken_requires_dedicated_USD_and_configured_asset_spot_account')

    async def snapshot(self):
        # Sample authenticated balance state and a fresh public bid. Private streams
        # avoid 1 Hz private REST polling that can exhaust Kraken's rate budget.
        if self.balance_ready.is_set():
            if time.monotonic() - self.last_private_message > 10:
                raise Fault('kraken_private_stream_stale')
            balances = dict(self.balance)
        else:
            balances = await self.rest('Balance')
        self.validate_balances(balances)
        cash = sum((decimal(v) for a, v in balances.items() if a in ('USD', 'ZUSD')), ZERO)
        qty = sum((decimal(v) for a, v in balances.items() if a not in ('USD', 'ZUSD')), ZERO)
        ticker = await self.rest('Ticker', {'pair': self.pair}, private=False)
        bid = decimal(next(iter(ticker.values()))['b'][0])
        if bid <= 0:
            raise Fault('invalid_kraken_bid')
        gross = qty * bid
        return {'at': now(), 'cash': cash, 'quantity': qty, 'equity': cash + gross,
                'gross_asset_value': gross, 'source': 'kraken_balances_and_public_bid',
                'positions':[{'symbol':self.symbol,'quantity':qty,'price':bid,
                    'cost':decimal(self.journal.get('asset_cost','0')),
                    'cost_basis':'fee_inclusive_authoritative_receipts'}] if qty>0 else []}

    async def orders(self):
        return list((await self.rest('OpenOrders'))['open'].values())

    async def buy_allowed(self):
        status = await self.rest('SystemStatus', private=False)
        if status.get('status') != 'online' or self.ws is None or self.ws.closed:
            raise Fault('kraken_exchange_or_authenticated_socket_unavailable')

    async def request_ws(self, method, params):
        self.require_auth(bound=True)
        if self.ws is None or self.ws.closed:
            raise Fault('kraken_authenticated_socket_unavailable')
        self.req += 1
        rid = self.req
        future = asyncio.get_running_loop().create_future()
        self.ws_pending[rid] = future
        try:
            await self.ws.send_str(wire({'method': method, 'params': {**params, 'token': self.token}, 'req_id': rid}))
            row = await asyncio.wait_for(future, timeout=8)
            if row.get('success') is not True:
                raise Fault('kraken_websocket_request_rejected')
            return row['result']
        finally:
            self.ws_pending.pop(rid, None)

    async def submit(self, cid, side, qty):
        self.require_auth(bound=True)
        qty = (qty / self.qty_step).to_integral_value(rounding=ROUND_DOWN) * self.qty_step
        if qty < self.minimum:
            raise Fault('kraken_order_below_minimum')
        result = await self.request_ws('add_order', {'order_type': 'market', 'side': side,
            'order_qty': qty, 'symbol': self.symbol, 'cl_ord_id': cid, 'fee_preference': 'quote',
            'deadline': (datetime.now(timezone.utc) + timedelta(seconds=5)).isoformat(timespec='milliseconds')})
        return result['order_id']

    async def find_order(self, cid):
        opened = await self.rest('OpenOrders', {'cl_ord_id': cid})
        closed = await self.rest('ClosedOrders', {'cl_ord_id': cid})
        matches = {**opened['open'], **closed['closed']}
        if len(matches) != 1:
            raise Fault('kraken_recovery_order_missing_or_ambiguous')
        return next(iter(matches))

    async def order_info(self, bid):
        rows = await self.rest('QueryOrders', {'txid': bid})
        if bid not in rows:
            raise Fault('kraken_order_reconciliation_missing')
        row = rows[bid]
        return {'broker_id': bid, 'status': row['status'], 'quantity': decimal(row['vol_exec']),
                'price': decimal(row['price']), 'fee': decimal(row['fee']),
                'fee_basis': 'authoritative_broker_order_aggregate', 'at': now()}

    async def cancel_all(self):
        await self.rest('CancelAll')

    async def bars(self):
        async with self.protected_ws(self.PUBLIC, heartbeat=20, max_msg_size=4 * 1024 * 1024) as ws:
            await ws.send_json({'method': 'subscribe', 'params': {'channel': 'ohlc',
                'symbol': [self.symbol], 'interval': 1, 'snapshot': False}})
            async for row in self.frames(ws):
                if row.get('success') is False:
                    raise Fault('kraken_market_subscription_rejected')
                if row.get('channel') == 'ohlc' and row.get('type') == 'update':
                    for bar in row['data']:
                        if bar['symbol'] == self.symbol:
                            yield {'at': bar['interval_begin'], 'price': decimal(bar['close']),
                                   'source': 'kraken_ohlc_update'}

    async def account_stream(self):
        async with self.protected_ws(self.PRIVATE, heartbeat=5, max_msg_size=4 * 1024 * 1024) as ws:
            self.ws = ws
            for channel in ('balances', 'executions'):
                await ws.send_json({'method': 'subscribe', 'params': {'channel': channel,
                    'token': self.token, **({'snapshot': True} if channel == 'balances' else {})}})
            try:
                async for row in self.frames(ws):
                    self.last_private_message = time.monotonic()
                    future = self.ws_pending.get(row.get('req_id'))
                    if future is not None and not future.done():
                        future.set_result(row)
                    if row.get('success') is False:
                        raise Fault('kraken_private_stream_rejected')
                    if row.get('channel') == 'balances':
                        sequence = row.get('sequence')
                        if row.get('type') == 'snapshot':
                            self.balance.clear()
                        elif self.sequence is not None and sequence != self.sequence + 1:
                            raise Fault('kraken_balance_sequence_gap')
                        self.sequence = sequence
                        for asset in row['data']:
                            for wallet in asset.get('wallets', []):
                                if decimal(wallet['balance']) != 0 and (wallet['type'] != 'spot' or wallet['id'] != 'main'):
                                    raise Fault('kraken_nonspot_wallet_balance')
                            if asset.get('wallet_type', 'spot') != 'spot':
                                raise Fault('kraken_nonspot_wallet_activity')
                            self.balance[asset['asset']] = decimal(asset['balance'])
                            if row.get('type') == 'update' and asset.get('type') not in ('trade',):
                                self.journal.halt('external_account_movement_' + asset.get('type', 'unknown'))
                        self.validate_balances(self.balance)
                        self.balance_ready.set()
                        self.stream_ready.set()
                    elif row.get('channel') == 'executions':
                        for event in row['data']:
                            self.journal.append('ORDER_STREAM', {key: event[key] for key in
                                ('order_id', 'cl_ord_id', 'exec_type', 'order_status', 'cum_qty', 'avg_price') if key in event})
            finally:
                self.ws = None
                self.balance_ready.clear()
                for future in list(self.ws_pending.values()):
                    if not future.done():
                        future.set_exception(Fault('kraken_socket_lost_during_request'))


class thread_gate:
    """Cancellation-safe acquisition of a cross-thread lock without blocking the loop."""
    def __init__(self, lock):
        self.lock = lock

    async def __aenter__(self):
        while not self.lock.acquire(blocking=False):
            await asyncio.sleep(0.01)
        return self

    async def __aexit__(self, *args):
        self.lock.release()


def broker(cfg, key, secret, journal):
    return (Alpaca if cfg['broker'] == 'alpaca' else Kraken)(cfg, key, secret, journal)


def account_mismatch(snapshot, journal):
    if not journal.get('initialized'):
        return False
    quantity = decimal(snapshot['quantity'])
    expected = decimal(journal.get('quantity', '0'))
    cash = decimal(snapshot['cash'])
    low = decimal(journal.get('cash_lower', journal.cfg['starting_capital']))
    high = decimal(journal.get('cash_upper', journal.cfg['starting_capital']))
    return abs(quantity - expected) > D('0.000000001') or cash < low - D('0.01') or cash > high + D('0.01')


def indicators(root, at):
    """Re-read monthly/event sources on every strategy tick; bounded file size.

    Empty/new project => HOLD, while networking and risk monitoring still run.
    Post-2000 supplied publication dates control visibility. Pre-2000 rows use
    observation dates under the owner's retained historical alignment policy.
    """
    months, events, future_in_month = {}, [], False
    for filename in ('macro.jsonl', 'geopolitical.jsonl'):
        path = root / filename
        if not path.is_file() or path.stat().st_size > 16 * 1024 * 1024:
            raise Fault('missing_or_oversized_' + filename)
        for line in path.read_text().splitlines():
            if not line.strip():
                continue
            row = decode(line)
            if filename == 'macro.jsonl':
                if row['series'] != 'INDPRO':
                    continue
                obs = timestamp(row['observation_date'] + ('T00:00:00+00:00' if len(row['observation_date']) == 10 else ''))
                supplied = row.get('available_at', row.get('publication_date'))
                available = obs if obs.year < 2000 or not supplied else timestamp(supplied)
                if available < obs:
                    raise Fault('macro_availability_before_observation')
                if obs <= at and available <= at:
                    month = obs.strftime('%Y-%m')
                    if month not in months or available > months[month][0]:
                        months[month] = (available, decimal(row['value']))
            else:
                event_at = timestamp(row['at'])
                if event_at.strftime('%Y-%m') == at.strftime('%Y-%m'):
                    if event_at <= at:
                        events.append(decimal(row['tone']))
                    else:
                        future_in_month = True
    keys = sorted(months)[-4:]
    indexes = [int(k[:4]) * 12 + int(k[5:]) for k in keys]
    values = [months[k][1] for k in keys] if len(keys) == 4 and indexes[-1] - indexes[0] == 3 else []
    tone = min(events) if events else (None if future_in_month else ZERO)
    return {'indpro': values, 'tone': tone, 'month': keys[-1] if keys else '',
            'macro_visibility_policy': 'supplied_available_at_else_observation_date'}


async def reconcile_pending(adapter, journal):
    with journal.mutex:
        pending = journal.db.execute("SELECT client_id,broker_id FROM intents WHERE status='pending'").fetchall()
    for cid, bid in pending:
        if not bid:
            try:
                bid = await adapter.find_order(cid)
                with journal.mutex, journal.db:
                    journal.db.execute('UPDATE intents SET broker_id=? WHERE client_id=?', (bid, cid))
                    journal._append('RECOVERED_ORDER_ID', {'client_id': cid, 'broker_id': bid})
            except Fault:
                journal.append('AMBIGUOUS_INTENT', {'client_id': cid, 'policy': 'never_resubmit'})
                continue
        if bid:
            try:
                row = await adapter.terminal(bid, timeout=10)
                journal.receipt(cid, row)
            except Fault:
                journal.append('RECONCILIATION_UNCONFIRMED', {'client_id': cid, 'broker_id': bid})
        else:
            journal.append('AMBIGUOUS_INTENT', {'client_id': cid, 'policy': 'never_resubmit'})


async def execute(adapter, journal, side, quantity, reference, month='', emergency=False):
    if quantity <= 0:
        raise Fault('nonpositive_order_quantity')
    adapter.require_auth(bound=True)
    cid = journal.intent(side, quantity, reference, month, emergency)
    # Persist the intent first; submission is NEVER retried after a timeout.
    bid = await adapter.submit(cid, side, quantity)
    with journal.mutex, journal.db:
        journal.db.execute('UPDATE intents SET broker_id=? WHERE client_id=?', (bid, cid))
        journal._append('ORDER_ACK', {'client_id': cid, 'broker_id': bid})
    row = await adapter.terminal(bid)
    journal.receipt(cid, row)
    expected = quantity
    if isinstance(adapter, Kraken):
        expected = (quantity / adapter.qty_step).to_integral_value(rounding=ROUND_DOWN) * adapter.qty_step
    if (row['status'] not in ('filled', 'closed') or decimal(row['quantity']) <= 0
            or abs(decimal(row['quantity']) - expected) > D('0.000000001')):
        raise Fault('entry_or_exit_not_fully_filled')
    return row


async def liquidate(adapter, journal, reason):
    """Cancel, reconcile, then close the managed position and verify flatness.

    Account scope errors and unsellable dust are UNCONFIRMED, never false success.
    Retrying this function reconciles the book first; pending submissions are not
    replayed. A failed/ambiguous liquidation is not automatically repeated.
    """
    journal.halt(reason)
    # Cancel immediately, independently of the strategy's fill-wait lock. This
    # unblocks terminal reconciliation before serializing emergency submissions.
    try:
        await adapter.cancel_all()
    except Exception as exc:
        journal.append('LIQUIDATION_UNCONFIRMED', {'reason': reason, 'stage': 'cancel_open_orders',
            'error_type': type(exc).__name__})
        log('LIQUIDATION_UNCONFIRMED', stage='cancel_open_orders')
        return False
    async with thread_gate(journal.order_gate):
        try:
            await reconcile_pending(adapter, journal)
            if await adapter.orders():
                raise Fault('cancel_not_reconciled')
            with journal.mutex:
                unresolved = journal.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0]
            with journal.mutex:
                pending_sells = journal.db.execute(
                    "SELECT body FROM intents WHERE status='pending'").fetchall()
            if any(decode(r[0])['side'] == 'sell' for r in pending_sells):
                raise Fault('prior_sell_ambiguous_no_duplicate_liquidation')
            # Unknown submission cannot be proven absent by open-orders alone: it
            # may have filled. Close observed inventory, retain unresolved audit state.
            snap = await adapter.snapshot()
            qty = decimal(snap['quantity'])
            if qty > 0:
                reference = decimal(snap['gross_asset_value']) / qty
                await execute(adapter, journal, 'sell', qty, reference, emergency=True)
            snap = await adapter.snapshot()
            if decimal(snap['quantity']) != 0 or await adapter.orders() or unresolved:
                raise Fault('flatness_or_prior_submission_unconfirmed')
            journal.observe('LIQUIDATION_CONFIRMED', snap, reason=reason)
            log('LIQUIDATION_CONFIRMED')
            return True
        except Exception as exc:
            journal.append('LIQUIDATION_UNCONFIRMED', {'reason': reason, 'error_type': type(exc).__name__,
                'fault': str(exc) if isinstance(exc, Fault) else 'internal_error'})
            log('LIQUIDATION_UNCONFIRMED', error_type=type(exc).__name__)
            return False


class RiskWorker(threading.Thread):
    """A separate event loop/session continues when the strategy loop stalls."""
    def __init__(self, cfg, key, secret, journal):
        super().__init__(name='alphabot-independent-risk', daemon=False)
        self.cfg, self.key, self.secret, self.journal = cfg, key, secret, journal
        self.stop_event = threading.Event()
        self.ready = threading.Event()
        self.last_sample = 0
        self.failed = False

    def run(self):
        getcontext().prec = 50
        try:
            asyncio.run(self.monitor())
        except BaseException:
            self.failed = True
            self.journal.halted.set()
            with contextlib.suppress(Exception):
                self.journal.halt('risk_worker_failed')
            self.ready.set()

    async def monitor(self):
        async with broker(self.cfg, self.key, self.secret, self.journal) as adapter:
            await adapter.prepare()
            # Kraken emergency submissions require its own authenticated socket.
            stream = asyncio.create_task(adapter.account_stream()) if isinstance(adapter, Kraken) else None
            if stream:
                await asyncio.wait_for(adapter.balance_ready.wait(), 10)
            due = time.monotonic()
            try:
                while not self.stop_event.is_set():
                    if stream and stream.done():
                        stream.result()
                        raise Fault('risk_private_stream_ended')
                    try:
                        generation = self.journal.get('execution_generation', 0)
                        snap = await adapter.snapshot()
                        self.last_sample = time.monotonic()
                        net, floor = self.journal.observe('RISK_SAMPLE', snap)
                        context = indicators(self.journal.root, datetime.now(timezone.utc))
                        self.ready.set()
                        with self.journal.mutex:
                            stable = generation == self.journal.get('execution_generation', 0) and not self.journal.order_gate.locked()
                            mismatch = stable and account_mismatch(snap, self.journal)
                        reason = ('account_reconciliation_mismatch' if mismatch else
                            'unclassified_accounting_movement' if self.journal.get('accounting_unresolved') else
                            'equity_below_floor' if net < floor else
                            'geopolitical_risk_override' if context['tone'] is not None and context['tone'] < -5 else
                            'durable_halt' if self.journal.halted.is_set() else None)
                        if reason:
                            await liquidate(adapter, self.journal, reason)
                            return
                    except Exception as exc:
                        self.ready.set()
                        await liquidate(adapter, self.journal, 'risk_sample_failed_' + type(exc).__name__)
                        self.failed = True
                        return
                    due += 1
                    delay = due - time.monotonic()
                    if delay < -1:
                        await liquidate(adapter, self.journal, 'risk_sampling_deadline_missed')
                        self.failed = True
                        return
                    await asyncio.sleep(max(0, delay))
            finally:
                if stream:
                    stream.cancel()
                    await asyncio.gather(stream, return_exceptions=True)


def security_settings(root, cfg):
    path = root / 'security.json'
    settings = decode(path.read_text()) if path.exists() else {}
    if settings.get('pin_gateway_origin') is True:
        if cfg.get('gateway_hostname') and not settings.get('gateway_hostname'):
            settings['gateway_hostname'] = cfg['gateway_hostname']
    else:
        # The administrator's configured HTTPS URL is authoritative. Legacy rc2
        # generated hostname defaults are not remote-edge preflight requirements.
        settings.pop('gateway_hostname', None)
        settings.pop('gateway_port', None)
    vault_path = os.environ.get('ALPHABOT_AUTH_VAULT') or settings.get('vault_path', 'auth.vault.json')
    settings['vault_path'] = str((root / Path(vault_path).expanduser()).resolve())
    settings.setdefault('account_id', cfg.get('account_id', ''))
    return settings


async def bind_broker_session(adapter, journal, key, secret):
    gate = journal.gate
    gate.require()
    account_id = journal.get('broker_account_id') if journal.cfg['broker']=='alpaca' else gate.settings.get('account_id')
    if gate.bound_account != account_id or not account_id:
        await gate.bootstrap(journal.cfg['broker'], account_id, key, secret)
    known_scope = journal.get('verified_broker_scope')
    scope = {'broker':journal.cfg['broker'],'account_id':account_id}
    if known_scope and known_scope != scope:
        raise Fault('verified_broker_scope_mismatch')
    if not known_scope and not journal.binding_mismatch:
        with journal.mutex, journal.db:
            journal.set('verified_broker_scope', scope)
            journal._append('BROKER_SCOPE_VERIFIED', scope)
    # Native signing needs in-memory credentials; remove inherited environment copies.
    os.environ.pop('LIVE_BROKER_API_KEY', None)
    os.environ.pop('LIVE_BROKER_SECRET_KEY', None)


async def prepare_diagnostics(journal):
    journal.gate.require()
    if journal.get('accounting_unresolved'):
        raise Fault('unclassified_accounting_blocks_strategy')
    if journal.observer is None or not journal.observer.attested(journal.root):
        raise Fault('connected_correct_database_observer_required')
    with journal.mutex:
        event_id = journal.db.execute('SELECT coalesce(max(id),0) FROM events').fetchone()[0]
    output = journal.root / ('live_diagnostics_' + str(event_id) + '_' + uuid.uuid4().hex[:8] + '.html')
    proof = await asyncio.to_thread(audit_viewer.diagnostic, journal.root/'native-live.sqlite3', output,
        {'event_id':event_id,'observer_database':'native-live.sqlite3','authorization_generation':journal.gate.generation})
    journal.append('DIAGNOSTIC_VERIFIED', {'event_id':event_id,'report':output.name,
        'tip_hash':proof['tip_hash'],'postings':proof['postings']})


async def trading(root, cfg, key, secret, journal, check=False, duration=None):
    if journal.binding_mismatch:
        raise Fault('binding_review_required_before_broker_execution')
    async with broker(cfg, key, secret, journal) as adapter:
        await adapter.prepare()
        await bind_broker_session(adapter, journal, key, secret)
        initial = await adapter.snapshot()
        journal.observe('PREFLIGHT', initial)
        if check:
            log('READ_ONLY_CHECK', snapshot=initial, halted=journal.halted.is_set(),
                unresolved_intents=journal.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0])
            return
        # A new project cannot import unknown historical acquisition cost silently.
        if not journal.get('initialized'):
            if decimal(initial['quantity']) != 0 or await adapter.orders():
                raise Fault('first_start_requires_flat_dedicated_account')
            if abs(decimal(initial['cash']) - decimal(cfg['starting_capital'])) > D('0.01'):
                raise Fault('starting_capital_does_not_match_account_cash')
            with journal.mutex, journal.db:
                journal.set('initialized', True)
                journal.set('quantity', '0')
                journal.set('asset_cost', '0')
                journal.set('cash_lower', cfg['starting_capital'])
                journal.set('cash_upper', cfg['starting_capital'])
        elif abs(decimal(initial['quantity']) - decimal(journal.get('quantity', '0'))) > D('0.000000001'):
            journal.halt('restart_holdings_reconciliation_mismatch')
        private = asyncio.create_task(adapter.account_stream(), name='authenticated-account-stream')
        risk = RiskWorker(cfg, key, secret, journal)
        risk.start()
        stopping = asyncio.Event()
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGINT, signal.SIGTERM):
            with contextlib.suppress(NotImplementedError):
                loop.add_signal_handler(sig, stopping.set)
        started = time.monotonic()
        async def consume():
            await asyncio.wait_for(adapter.stream_ready.wait(), 10)
            deadline = time.monotonic() + 15
            while not risk.ready.is_set():
                if time.monotonic() >= deadline:
                    raise Fault('risk_worker_startup_timeout')
                await asyncio.sleep(0.01)
            async for bar in adapter.bars():
                price = decimal(bar['price'])
                age = D(str(time.time())) - D(str(timestamp(bar['at']).timestamp()))
                if price <= 0 or age < -5 or age > cfg['market_data_max_age_seconds']:
                    raise Fault('stale_future_or_invalid_bar')
                snap = await adapter.snapshot()
                net, floor = journal.observe('MARKET_TICK', snap, bar=bar)
                if journal.halted.is_set() or stopping.is_set():
                    return
                if net < floor:
                    await liquidate(adapter, journal, 'strategy_preflight_floor_breach')
                    return
                context = {**indicators(root, datetime.now(timezone.utc)), 'cash': snap['cash'],
                    'last_entry_month': journal.get('last_entry_month'),
                    'lesson_count': journal.lesson_count(), 'lesson_risk_decay': cfg['lesson_risk_decay']}
                await journal.gate.renew_if_due()
                journal.gate.require()
                if journal.observer is None or not journal.observer.attested(journal.root):
                    journal.append('ENTRY_BLOCKED', {'reason':'connected_correct_database_observer_required'})
                    continue
                await prepare_diagnostics(journal)
                decision = decide(context)
                journal.append('DECISION', {'context': context, 'decision': decision})
                if decision['action'] == 'liquidate':
                    await liquidate(adapter, journal, decision['reason'])
                    return
                if decision['action'] != 'buy':
                    continue
                async with thread_gate(journal.order_gate):
                    if journal.halted.is_set() or risk.failed or time.monotonic() - risk.last_sample > 3:
                        raise Fault('risk_health_blocks_order')
                    await adapter.buy_allowed()
                    if await adapter.orders():
                        raise Fault('unmanaged_open_orders')
                    fresh = await adapter.snapshot()
                    if account_mismatch(fresh, journal):
                        raise Fault('account_movement_or_holdings_mismatch')
                    net, floor = journal.observe('ORDER_PREFLIGHT', fresh)
                    budget = min(decimal(decision['budget']), decimal(cfg['max_order_usd']), decimal(fresh['cash']))
                    drag = decimal(cfg['fee_reserve_rate']) + decimal(cfg['slippage_reserve_rate'])
                    # Reserve modeled entry+exit costs; market fills can still gap.
                    projected_floor, _ = capital_floor(journal.get('protected_starting_capital', cfg['starting_capital']),
                        decimal(journal.get('asset_cost', '0')) + budget,
                        journal.get('peak', cfg['starting_capital']), journal.get('ratchet', '250'))
                    if net - budget * drag * 2 < projected_floor:
                        journal.append('ORDER_BLOCKED', {'reason': 'projected_floor', 'budget': budget})
                        continue
                    qty = (budget / (price * (1 + drag))).quantize(D('0.00000001'), rounding=ROUND_DOWN)
                    if qty <= 0 or (isinstance(adapter, Kraken) and (qty * price < adapter.cost_minimum or qty < adapter.minimum)):
                        journal.append('ORDER_BLOCKED', {'reason': 'venue_minimum_notional'})
                        continue
                    journal.gate.require()
                    await prepare_diagnostics(journal)
                    await execute(adapter, journal, 'buy', qty, price, context['month'])
        consumer = asyncio.create_task(consume(), name='live-market-execution')
        try:
            while not stopping.is_set():
                await journal.gate.renew_if_due()
                if private.done():
                    private.result()
                    raise Fault('account_stream_ended')
                if consumer.done():
                    consumer.result()
                    break
                if not risk.is_alive() or risk.failed:
                    break
                if duration is not None and time.monotonic() - started >= duration:
                    break
                await asyncio.sleep(0.1)
        except Exception as exc:
            journal.halt('execution_failure_' + type(exc).__name__)
            log('EXECUTION_FAILURE', error_type=type(exc).__name__,
                fault=str(exc) if isinstance(exc, Fault) else 'internal_error')
        finally:
            # Stop admission before canceling a possibly in-flight strategy request.
            journal.halt('session_shutdown')
            consumer.cancel()
            await asyncio.gather(consumer, return_exceptions=True)
            risk.stop_event.set()
            # Risk gets first ownership of emergency reconciliation if already active.
            while risk.is_alive():
                await asyncio.sleep(0.05)
            await liquidate(adapter, journal, 'session_shutdown')
            private.cancel()
            await asyncio.gather(private, return_exceptions=True)
            journal.verify()


async def rearm(cfg, key, secret, journal, reason):
    if journal.binding_mismatch:
        raise Fault('binding_review_required_before_rearm')
    async with broker(cfg, key, secret, journal) as adapter:
        await adapter.prepare()
        await bind_broker_session(adapter, journal, key, secret)
        await reconcile_pending(adapter, journal)
        snapshot = await adapter.snapshot()
        net, floor = journal.observe('REARM_PREFLIGHT', snapshot)
        with journal.mutex:
            unresolved = journal.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0]
        if (unresolved or await adapter.orders() or decimal(snapshot['quantity']) != 0
                or account_mismatch(snapshot, journal) or net < floor or journal.get('accounting_unresolved')):
            raise Fault('rearm_requires_reconciled_flat_account_above_floor')
        with journal.mutex, journal.db:
            old_halt = journal.get('halt')
            journal._append('OPERATOR_REARM', {'reason': reason, 'prior_halt': old_halt,
                'preserved_peak': journal.get('peak'), 'preserved_ratchet': journal.get('ratchet'),
                'preserved_lesson_count': journal.lesson_count()})
            journal.set('halt', None)
            journal.halted.clear()
        log('REARMED', preserved_lesson_count=journal.lesson_count())


def initialize(root, name, symbol):
    root.mkdir(parents=True, exist_ok=True, mode=0o700)
    if (root / 'config.json').exists():
        raise Fault('refusing_to_overwrite_existing_project')
    cfg = {'broker': name, 'symbol': symbol or ('SPY' if name == 'alpaca' else 'BTC/USD'),
           'starting_capital': '500', 'max_order_usd': '100', 'fee_reserve_rate': '0.005',
           'slippage_reserve_rate': '0.01', 'lesson_risk_decay': '0.95',
           'execution_target_multiplier': '1', 'dedicated_account': True}
    (root / 'config.json').write_text(json.dumps(cfg, indent=2) + '\n')
    (root / 'security.json').write_text(json.dumps({'vault_path':'auth.vault.json',
        'account_id':''}, indent=2) + '\n')
    (root / 'macro.jsonl').touch(exist_ok=False)
    (root / 'geopolitical.jsonl').touch(exist_ok=False)
    log('PROJECT_INITIALIZED', project=str(root), policy='empty_INDPRO_means_HOLD')


async def wait_for_observer_stop(duration=None):
    stopped = asyncio.Event()
    loop = asyncio.get_running_loop()
    if threading.current_thread() is threading.main_thread():
        for sig in (signal.SIGINT, signal.SIGTERM):
            with contextlib.suppress(NotImplementedError):
                loop.add_signal_handler(sig, stopped.set)
    try:
        if duration is None:
            await stopped.wait()
        else:
            await asyncio.wait_for(stopped.wait(), timeout=duration)
    except asyncio.TimeoutError:
        return


async def approve_local_binding(cfg, key, secret, journal, reason):
    """Explicit authenticated, flat-account metadata migration; never reset risk."""
    journal.gate.require()
    if not journal.binding_mismatch:
        log('BINDING_CURRENT')
        return
    with journal.mutex:
        pending = journal.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0]
    if (pending or decimal(journal.get('quantity','0')) != 0 or
            decimal(journal.get('asset_cost','0')) != 0 or journal.get('accounting_unresolved')):
        raise Fault('binding_review_requires_reconciled_flat_ledger')
    known = journal.get('verified_broker_scope')
    if known and known['broker'] != cfg['broker']:
        raise Fault('binding_review_cannot_change_broker')
    if not known and cfg['broker']=='kraken' and journal.get('initialized'):
        raise Fault('legacy_Kraken_scope_requires_verified_migration')
    async with broker(cfg,key,secret,journal) as adapter:
        await adapter.prepare()
        await bind_broker_session(adapter,journal,key,secret)
        account_id = journal.get('broker_account_id') if cfg['broker']=='alpaca' else journal.gate.settings.get('account_id')
        if known and known['account_id'] != account_id:
            raise Fault('binding_review_cannot_change_account')
        snapshot = await adapter.snapshot()
        if decimal(snapshot['quantity']) != 0 or await adapter.orders() or account_mismatch(snapshot,journal):
            raise Fault('binding_review_requires_same_reconciled_flat_account')
        previous_floor = decimal(cfg['starting_capital'])
        with journal.mutex:
            row = journal.db.execute("SELECT body FROM events WHERE kind IN ('PREFLIGHT','RISK_SAMPLE','ORDER_PREFLIGHT','MARKET_TICK','LIQUIDATION_CONFIRMED') ORDER BY id DESC LIMIT 1").fetchone()
            if row:
                previous_floor = max(previous_floor,decimal(decode(row[0]).get('floor','250')))
        with journal.mutex,journal.db:
            old = journal.get('binding')
            journal.set('protected_starting_capital',max(previous_floor,decimal(journal.get('protected_starting_capital','0'))))
            journal.set('binding',journal.proposed_binding)
            journal.set('verified_broker_scope',{'broker':cfg['broker'],'account_id':account_id})
            journal._append('LOCAL_BINDING_REVIEW_APPROVED',{'old_binding':old,'new_binding':journal.proposed_binding,
                'reason':reason,'preserved_peak':journal.get('peak'),'preserved_ratchet':journal.get('ratchet'),
                'preserved_halt':journal.get('halt'),'preserved_lesson_count':journal.lesson_count()})
            journal.binding_mismatch=False
        log('LOCAL_BINDING_REVIEW_APPROVED',halt_preserved=journal.halted.is_set())


def arguments(argv=None):
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command',nargs='?',choices=('init','enroll-vault','auth-check','check','run','verify','rearm','rebind','observe'))
    parser.add_argument('--project',type=Path,default=Path.cwd(),help='selected project; defaults to the current directory')
    parser.add_argument('--broker',choices=('alpaca','kraken'),default='alpaca')
    parser.add_argument('--symbol')
    parser.add_argument('--live',action='store_true',help='enable a real-money session after authentication')
    parser.add_argument('--serve',action='store_true',help='read-only portal alone, or portal alongside --live')
    parser.add_argument('--acknowledge-halt',action='store_true')
    parser.add_argument('--acknowledge-binding-change',action='store_true')
    parser.add_argument('--reason')
    parser.add_argument('--port',type=int,default=8080)
    parser.add_argument('--duration',type=int)
    args=parser.parse_args(argv)
    if args.command is None:
        args.command='run' if args.live else 'observe' if args.serve else None
    if args.command is None:
        parser.error('supply a command, --serve or --live')
    if args.live and args.command!='run':
        parser.error('--live is valid only for run')
    if args.serve and args.command not in ('run','observe'):
        parser.error('--serve is valid only for run or observe')
    return args


def enroll_vault(root):
    if not sys.stdin.isatty() or threading.current_thread() is not threading.main_thread():
        raise AuthenticationFailed()
    resource.setrlimit(resource.RLIMIT_CORE,(0,0))
    password=getpass.getpass(AdministratorGate.PROMPT)
    confirm=getpass.getpass('Confirm App-Level Secret Password: ')
    if not hmac.compare_digest(password.encode(),confirm.encode()):
        raise AuthenticationFailed()
    seed=getpass.getpass('Enter Existing Google Authenticator Base32 Secret: ')
    token=getpass.getpass(AdministratorGate.MFA_PROMPT)
    counter=int(time.time())//30
    if not re.fullmatch(r'[0-9]{6}',token) or not any(
            hmac.compare_digest(LocalVault.token(seed,n),token) for n in (counter-1,counter,counter+1)):
        raise AuthenticationFailed()
    settings=security_settings(root,{})
    vault=LocalVault(settings['vault_path'])
    vault.create(password,seed)
    vault.verify(password,token)
    password=confirm=seed=token=None
    log('LOCAL_VAULT_ENROLLED',next_login_requires_unused_Authenticator_code=True)


def main(argv=None):
    args=arguments(argv)
    # First application action for a valid --live invocation: both prompts.
    # No config read, journal binding, credential lookup or network precedes them.
    factors=AdministratorGate.prompt_factors() if (args.live or args.command in ('check','rearm','rebind','auth-check')) else None
    try:
        root=args.project.expanduser().resolve()
        if args.command=='init':
            initialize(root,args.broker,args.symbol)
            return 0
        if args.command=='enroll-vault':
            enroll_vault(root)
            return 0
        if args.command=='auth-check':
            settings=security_settings(root,{})
            LocalVault(settings['vault_path']).verify(*factors)
            factors=None
            log('LOCAL_FACTORS_VERIFIED',gateway_session=False,broker_execution=False)
            return 0
        if not 1<=args.port<=65535 or (args.duration is not None and args.duration<=0):
            raise Fault('invalid_port_or_duration')
        if args.command=='verify':
            db=audit_viewer.open_readonly(root/'native-live.sqlite3')
            try:log('JOURNAL_VERIFIED',**audit_viewer.verify(db))
            finally:db.close()
            return 0
        if args.command=='observe':
            async def observe_only():
                observer=await Observer(root,args.port).start()
                log('OBSERVER',url=observer.endpoint,read_only=True)
                try:await wait_for_observer_stop(args.duration)
                finally:await observer.close()
            asyncio.run(observe_only())
            return 0
        if args.command=='run' and not args.live:
            raise Fault('run_requires_explicit_--live')
        if args.command=='rearm' and (not args.acknowledge_halt or not args.reason or len(args.reason)>200):
            raise Fault('rearm_requires_--acknowledge-halt_and_--reason_up_to_200_characters')
        if args.command=='rebind' and (not args.acknowledge_binding_change or not args.reason or len(args.reason)>200):
            raise Fault('rebind_requires_explicit_acknowledgment_and_reason')
        cfg=configuration(root)
        async def secured():
            nonlocal factors
            gate=AdministratorGate(security_settings(root,cfg),os.environ)
            journal=observer=None
            try:
                await gate.open(factors)
                factors=None
                key=os.environ.get('LIVE_BROKER_API_KEY')
                secret=os.environ.get('LIVE_BROKER_SECRET_KEY')
                if not key or not secret:
                    raise Fault('LIVE_BROKER_API_KEY_and_LIVE_BROKER_SECRET_KEY_required')
                journal=Journal(root,cfg,hashlib.sha256(key.encode()).hexdigest(),defer_binding_review=True)
                journal.gate=gate
                gate.on_failure=journal.halt
                if args.command=='run':
                    observer=await Observer(root,args.port).start()
                    journal.observer=observer
                    log('OBSERVER',url=observer.endpoint,read_only=True)
                if journal.binding_mismatch and args.command!='rebind':
                    log('LOCAL_BINDING_REVIEW_REQUIRED',remote_preflight=False,
                        instructions='Use authenticated rebind with acknowledgment and reason; existing risk state is preserved')
                    if observer:
                        await wait_for_observer_stop(args.duration)
                    return 2
                if args.command=='rebind':
                    await approve_local_binding(cfg,key,secret,journal,args.reason)
                elif args.command=='rearm':
                    await rearm(cfg,key,secret,journal,args.reason)
                else:
                    await trading(root,cfg,key,secret,journal,check=args.command=='check',duration=args.duration)
                return 2 if journal.halted.is_set() else 0
            finally:
                if observer:await observer.close()
                await gate.close()
                if journal:journal.close()
        return asyncio.run(secured())
    finally:
        factors=None


if __name__=='__main__':
    try:
        raise SystemExit(main())
    except (Fault,AuthenticationFailed,ValueError,KeyError,OSError) as exc:
        log('STARTUP_FAILED',error_type=type(exc).__name__,
            fault=str(exc) if isinstance(exc,(Fault,AuthenticationFailed)) else 'invalid_input_or_system_error')
        raise SystemExit(2)