"""SYNTHETIC protocol fixtures: real local HTTP/WebSockets, never broker certification."""
import asyncio
import base64
from datetime import datetime, timezone
from decimal import Decimal as D
import hashlib
import hmac
import json
from pathlib import Path
import sqlite3
import sys
import os
import subprocess
import tempfile
import threading
import time
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import gauntlet_v2 as g
import host_strategy as h
import aiohttp
from aiohttp import web
from audit_dashboard import Observer


class FixtureGate:
    """Explicit synthetic gate injected ONLY by tests, never a runtime option."""
    settings = {'account_id':'synthetic-account'}
    bound_account = 'synthetic-account'
    generation = 1
    def require(self): return None
    async def renew_if_due(self): return None
    async def bootstrap(self, broker, account_id, key, secret): self.bound_account=account_id


class PolicyTests(unittest.TestCase):
    def test_floor_milestones_and_permanent_ratchet(self):
        for peak, expected in [('999', '250'), ('1000', '500'), ('1999', '500'),
                               ('2000', '1000'), ('4000', '2000'), ('8000', '4000')]:
            floor, ratchet = h.capital_floor('500', '500', peak)
            self.assertEqual(floor, D(expected))
            self.assertEqual(ratchet, D(expected))
        self.assertEqual(h.capital_floor('500', '500', '600', '2000')[0], D('2000'))

    def test_acquisition_cost_protection(self):
        self.assertEqual(h.capital_floor('500', '101', '500')[0], D('399'))

    def test_decimal_only(self):
        for value in (1.2, True, 'NaN', 'Infinity'):
            with self.assertRaises((ValueError, TypeError)):
                h.decimal(value)

    def test_threshold_and_cumulative_penalty(self):
        _, deviation = h.target_deviation('100', '100.0001')
        self.assertEqual(deviation, D('0.0001'))
        self.assertEqual(h.risk_multiplier(3), D('.95') ** 3)

    def test_strategy_uses_lessons(self):
        ctx = {'tone': '0', 'indpro': ['10', '11', '12', '20'], 'cash': '500',
            'month': '2026-10', 'last_entry_month': None, 'lesson_count': 3, 'lesson_risk_decay': '.95'}
        self.assertEqual(h.decide(ctx)['budget'], D('100') * D('.95') ** 3)
        ctx['tone'] = '-5.0001'
        self.assertEqual(h.decide(ctx)['action'], 'liquidate')

    def test_empty_indicators_hold(self):
        self.assertEqual(h.decide({'tone': '0', 'indpro': []})['action'], 'hold')

    def test_wire_preserves_decimal_numbers(self):
        value = D('123.123456789123456789')
        encoded = g.wire({'order_qty': value})
        self.assertNotIn('"123.', encoded)
        self.assertEqual(g.decode(encoded)['order_qty'], value)

    def test_accounting_rejects_binary_float(self):
        with self.assertRaises(ValueError):
            g.encode({'cash': 0.1})


class JournalTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        g.initialize(self.root, 'alpaca', 'SPY')
        self.cfg = g.configuration(self.root)
        self.j = g.Journal(self.root, self.cfg, 'synthetic-key-fingerprint')

    def tearDown(self):
        self.j.close()
        self.temp.cleanup()

    def test_receipt_lesson_and_idempotence(self):
        cid = self.j.intent('buy', D('1'), D('100'), '2026-10')
        receipt = {'broker_id': 'order-1', 'status': 'filled', 'quantity': D('1'),
            'price': D('100.0001'), 'fee': D('.1'), 'fee_basis': 'synthetic', 'at': g.now()}
        self.j.receipt(cid, receipt)
        self.j.receipt(cid, receipt)
        self.assertEqual(self.j.lesson_count(), 1)
        self.assertEqual(D(self.j.get('asset_cost')), D('100.1001'))
        self.assertEqual(self.j.get('last_entry_month'), '2026-10')
        self.j.verify()

    def test_halt_persists_and_blocks_entry(self):
        self.j.halt('test')
        with self.assertRaises(g.Fault):
            self.j.intent('buy', D('1'), D('100'))
        self.j.close()
        self.j = g.Journal(self.root, self.cfg, 'synthetic-key-fingerprint')
        self.assertTrue(self.j.halted.is_set())

    def test_pending_restart_never_resubmits(self):
        self.j.intent('buy', D('1'), D('100'))
        self.j.close()
        self.j = g.Journal(self.root, self.cfg, 'synthetic-key-fingerprint')
        self.assertTrue(self.j.halted.is_set())
        self.assertEqual(self.j.db.execute('SELECT count(*) FROM intents').fetchone()[0], 1)

    def test_scope_binding(self):
        self.j.close()
        with self.assertRaises(g.Fault):
            g.Journal(self.root, self.cfg, 'different-key')
        self.j = g.Journal(self.root, self.cfg, 'synthetic-key-fingerprint')

    def test_hash_tamper_detection(self):
        self.j.append('TEST', {'cash': D('1.00000001')})
        with self.assertRaises(sqlite3.IntegrityError):
            with self.j.db:
                self.j.db.execute("UPDATE events SET body='{}'")
        with self.j.db:
            self.j.db.execute('DROP TRIGGER native_immutable_events_UPDATE')
            self.j.db.execute("UPDATE events SET body='{}'")
        with self.assertRaises(g.Fault):
            self.j.verify()

    def test_every_observation_is_fixed_point_and_floor_survives(self):
        snap = {'equity': D('2000'), 'cash': D('2000'), 'quantity': D('0'), 'gross_asset_value': D('0')}
        self.j.observe('RISK_SAMPLE', snap)
        snap['equity'], snap['cash'] = D('600'), D('600')
        net, floor = self.j.observe('MARKET_TICK', snap)
        self.assertEqual(floor, D('1000'))
        row = self.j.db.execute("SELECT body FROM events WHERE kind='MARKET_TICK'").fetchone()[0]
        self.assertIsInstance(json.loads(row)['snapshot']['cash'], str)
        self.assertEqual(net, D('600'))

    def test_nonce_unique_across_threads(self):
        adapter = g.Kraken({**self.cfg, 'symbol': 'BTC/USD'}, 'synthetic', base64.b64encode(b'x').decode(), self.j)
        bodies = []
        def sign():
            bodies.append(adapter.signed('/0/private/Balance', {})[0])
        threads = [threading.Thread(target=sign) for _ in range(20)]
        for t in threads: t.start()
        for t in threads: t.join()
        self.assertEqual(len(set(bodies)), 20)

    def test_cli_verify_does_not_need_credentials(self):
        self.j.append('SYNTHETIC_CHECK', {'cash': D('500')})
        env = {k:v for k,v in os.environ.items() if k not in ('LIVE_BROKER_API_KEY','LIVE_BROKER_SECRET_KEY')}
        result = subprocess.run([sys.executable, str(Path(g.__file__)), 'verify', '--project', str(self.root)],
            env=env, capture_output=True, text=True, timeout=10)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assertIn('JOURNAL_VERIFIED', result.stdout)

    def test_cli_run_requires_explicit_live_flag(self):
        result = subprocess.run([sys.executable, str(Path(g.__file__)), 'run', '--project', str(self.root)],
            capture_output=True, text=True, timeout=10)
        self.assertEqual(result.returncode, 2)
        self.assertIn('run_requires_explicit_--live', result.stdout)

    def test_macro_visibility_and_month_gaps(self):
        (self.root / 'macro.jsonl').write_text('\n'.join(json.dumps({'series':'INDPRO',
            'observation_date': f'2026-{m:02d}-01', 'available_at': f'2026-{m:02d}-15T00:00:00+00:00',
            'value': str(m)}) for m in (6, 7, 8, 9)))
        result = g.indicators(self.root, datetime(2026, 9, 10, tzinfo=timezone.utc))
        self.assertEqual(result['indpro'], [])
        result = g.indicators(self.root, datetime(2026, 10, 1, tzinfo=timezone.utc))
        self.assertEqual(result['indpro'], [D(6),D(7),D(8),D(9)])


class ProtocolTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        g.initialize(self.root, 'alpaca', 'SPY')
        self.cfg = g.configuration(self.root)
        self.j = g.Journal(self.root, self.cfg, 'synthetic')
        self.cash, self.qty, self.price = D('500'), D('0'), D('100')
        self.orders = {}
        self.submissions = []
        self.force_equity = None
        self.trigger_breach = False
        self.order_behavior = 'filled'
        self.account_reads = []
        self.kraken_clients = {}
        app = web.Application()
        app.router.add_route('*', '/{path:.*}', self.handle)
        self.runner = web.AppRunner(app)
        await self.runner.setup()
        self.site = web.TCPSite(self.runner, '127.0.0.1', 0)
        await self.site.start()
        port = self.site._server.sockets[0].getsockname()[1]
        self.http = f'http://127.0.0.1:{port}'
        self.ws = f'ws://127.0.0.1:{port}'
        self.patches = [patch.object(g.Alpaca,'REST',self.http), patch.object(g.Alpaca,'DATA',self.ws+'/bars/'),
            patch.object(g.Alpaca,'ORDERS',self.ws+'/stream'), patch.object(g.Kraken,'REST',self.http),
            patch.object(g.Kraken,'PUBLIC',self.ws+'/kraken-public'), patch.object(g.Kraken,'PRIVATE',self.ws+'/kraken-private')]
        for p in self.patches: p.start()
        await self.start_observer()

    async def asyncTearDown(self):
        await self.stop_observer()
        await self.runner.cleanup()
        for p in self.patches: p.stop()
        self.j.close()
        self.temp.cleanup()

    async def start_observer(self):
        if hasattr(self, 'observer'):
            await self.stop_observer()
        self.j.gate = FixtureGate()
        self.observer = await Observer(self.root,0).start()
        self.j.observer = self.observer
        self.observer_client = aiohttp.ClientSession()
        self.observer_ws = await self.observer_client.ws_connect(self.observer.endpoint+'/ws')
        async def drain():
            async def beat():
                while not self.observer_ws.closed:
                    await self.observer_ws.send_json({'type':'heartbeat'})
                    await asyncio.sleep(.5)
            task=asyncio.create_task(beat())
            try:
                async for msg in self.observer_ws: pass
            finally:
                task.cancel()
                await asyncio.gather(task,return_exceptions=True)
        self.observer_drain = asyncio.create_task(drain())
        await asyncio.sleep(.01)

    async def stop_observer(self):
        self.observer_drain.cancel()
        await asyncio.gather(self.observer_drain,return_exceptions=True)
        await self.observer_ws.close()
        await self.observer_client.close()
        await self.observer.close()

    async def handle(self, request):
        path = request.path
        if path == '/stream':
            ws = web.WebSocketResponse()
            await ws.prepare(request)
            async for message in ws:
                row = json.loads(message.data)
                if row['action']=='auth':
                    await ws.send_json({'stream':'authorization','data':{'status':'authorized'}})
                else:
                    await ws.send_json({'stream':'listening','data':{'streams':['trade_updates']}})
            return ws
        if path.startswith('/bars/'):
            ws = web.WebSocketResponse()
            await ws.prepare(request)
            async for message in ws:
                row = json.loads(message.data)
                if row['action']=='auth':
                    await ws.send_json([{'T':'success','msg':'authenticated'}])
                else:
                    await ws.send_json([{'T':'subscription','bars':['SPY']}])
                    await asyncio.sleep(.1)
                    await ws.send_str(g.wire([{'T':'b','S':'SPY','t':g.now(),'c':self.price}]))
            return ws
        if path == '/v2/account':
            self.account_reads.append(time.monotonic())
            equity = self.force_equity if self.force_equity is not None else self.cash + self.qty*self.price
            return web.json_response({'id':'synthetic-account','currency':'USD','status':'ACTIVE',
                'trading_blocked':False,'multiplier':'1','cash':str(self.cash),'equity':str(equity),'short_market_value':'0'})
        if path == '/v2/assets/SPY':
            return web.json_response({'tradable':True,'fractionable':True})
        if path == '/v2/positions':
            return web.json_response([{'symbol':'SPY','side':'long','qty':str(self.qty),
                'market_value':str(self.qty*self.price)}] if self.qty else [])
        if path == '/v2/clock':
            return web.json_response({'is_open':True})
        if path == '/v2/orders:by_client_order_id':
            cid = request.query['client_order_id']
            for row in self.orders.values():
                if row.get('client_order_id') == cid:
                    return web.json_response(row)
            return web.json_response({}, status=404)
        if path == '/v2/orders':
            if request.method == 'GET':
                return web.json_response([v for v in self.orders.values() if v['status'] not in g.TERMINAL])
            if request.method == 'DELETE':
                for row in self.orders.values():
                    if row['status'] not in g.TERMINAL: row['status']='canceled'
                return web.json_response([])
            order = await request.json()
            self.submissions.append(order)
            qty = D(order['qty'])
            if self.order_behavior in ('partial', 'working_partial') and order['side']=='buy': qty /= 2
            if order['side']=='buy':
                self.qty += qty
                self.cash -= qty*self.price
                if self.trigger_breach:
                    asyncio.get_running_loop().call_later(.3, setattr, self, 'force_equity', D('200'))
            else:
                self.qty -= qty
                self.cash += qty*self.price
            bid = f'order-{len(self.orders)+1}'
            row = {'id':bid,'client_order_id':order['client_order_id'],
                'status':('canceled' if self.order_behavior=='partial' else 'new') if self.order_behavior in ('partial','working_partial') and order['side']=='buy' else 'filled',
                'filled_qty':str(qty),'filled_avg_price':str(self.price)}
            self.orders[bid] = row
            return web.json_response(row)
        if path.startswith('/v2/orders/'):
            return web.json_response(self.orders[path.rsplit('/',1)[1]])
        if path.startswith('/0/'):
            return await self.kraken_rest(request)
        if path in ('/kraken-public','/kraken-private'):
            return await self.kraken_socket(request)
        return web.json_response({}, status=404)

    async def kraken_rest(self, request):
        endpoint = request.path.rsplit('/',1)[1]
        if endpoint=='AssetPairs':
            result={'XXBTZUSD':{'wsname':'XBT/USD','status':'online','base':'XXBT','lot_decimals':8,'ordermin':'.0001','costmin':'.5'}}
        elif endpoint=='GetWebSocketsToken': result={'token':'SYNTHETIC-TOKEN'}
        elif endpoint=='OpenPositions': result={}
        elif endpoint=='SystemStatus': result={'status':'online'}
        elif endpoint=='Balance': result={'ZUSD':str(self.cash),'XXBT':str(self.qty)}
        elif endpoint=='Ticker': result={'XXBTZUSD':{'b':[str(self.price)]}}
        elif endpoint=='OpenOrders': result={'open':{}}
        elif endpoint=='CancelAll': result={'count':0}
        elif endpoint=='QueryOrders':
            form=await request.post()
            bid=form['txid']
            row=self.orders[bid]
            result={bid:{'status':'closed','vol_exec':row['qty'],'price':str(self.price),'fee':'.01'}}
        else: return web.json_response({'error':['unknown-fixture-endpoint']})
        if '/private/' in request.path:
            form = await request.post()
            body=await request.text()
            digest=hashlib.sha256((form['nonce']+body).encode()).digest()
            signature=base64.b64encode(hmac.new(b'synthetic-secret', request.path.encode()+digest,hashlib.sha512).digest()).decode()
            if signature != request.headers.get('API-Sign'):
                return web.json_response({'error':['invalid-signature']})
        return web.json_response({'error':[],'result':result})

    async def kraken_socket(self, request):
        ws=web.WebSocketResponse()
        await ws.prepare(request)
        async for message in ws:
            row=g.decode(message.data)
            if row['method']=='subscribe':
                channel=row['params']['channel']
                await ws.send_json({'method':'subscribe','success':True,'result':{'channel':channel}})
                if channel=='balances':
                    self.kraken_clients[ws] = 1
                    await ws.send_str(g.wire({'channel':'balances','type':'snapshot','sequence':1,
                        'data':[{'asset':'USD','balance':self.cash},{'asset':'BTC','balance':self.qty}]}))
                if channel=='ohlc':
                    await ws.send_str(g.wire({'channel':'ohlc','type':'update',
                        'data':[{'symbol':'BTC/USD','interval_begin':g.now(),'close':self.price}]}))
            elif row['method']=='add_order':
                params=row['params']
                self.assertIsInstance(params['order_qty'], D)
                self.assertEqual(params['token'],'SYNTHETIC-TOKEN')
                qty=params['order_qty']
                if params['side']=='buy': self.qty+=qty; self.cash-=qty*self.price+D('.01')
                else: self.qty-=qty; self.cash+=qty*self.price-D('.01')
                bid=f'kraken-{len(self.orders)+1}'
                self.orders[bid]={'qty':str(qty)}
                self.submissions.append(g.fixed(params))
                for client in list(self.kraken_clients):
                    if client.closed:
                        self.kraken_clients.pop(client, None)
                        continue
                    self.kraken_clients[client] += 1
                    await client.send_str(g.wire({'channel':'balances','type':'update',
                        'sequence':self.kraken_clients[client], 'data':[
                            {'asset':'USD','balance':self.cash,'type':'trade','wallet_type':'spot'},
                            {'asset':'BTC','balance':self.qty,'type':'trade','wallet_type':'spot'}]}))
                await ws.send_json({'method':'add_order','success':True,'req_id':row['req_id'],'result':{'order_id':bid}})
        return ws

    def monthly_signal(self):
        today=datetime.now(timezone.utc)
        index=today.year*12+today.month-1
        rows=[]
        for delta,value in zip((3,2,1,0),(10,11,12,20)):
            year,month=divmod(index-delta,12)
            rows.append({'series':'INDPRO','observation_date':f'{year:04d}-{month+1:02d}-01','value':str(value)})
        (self.root/'macro.jsonl').write_text('\n'.join(json.dumps(r) for r in rows)+'\n')

    async def test_alpaca_native_buy_risk_breach_liquidates(self):
        self.monthly_signal()
        self.trigger_breach=True
        await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,duration=4)
        self.assertEqual([o['side'] for o in self.submissions],['buy','sell'])
        self.assertEqual(self.qty,0)
        self.assertTrue(self.j.halted.is_set())
        kinds=[r[0] for r in self.j.db.execute('SELECT kind FROM events')]
        for kind in ('MARKET_TICK','RISK_SAMPLE','ORDER_INTENT','EXECUTION_RECEIPT','LIQUIDATION_CONFIRMED'):
            self.assertIn(kind,kinds)
        self.j.verify()

    async def test_partial_fill_is_accounted_and_halts(self):
        self.monthly_signal()
        self.order_behavior='partial'
        await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,duration=3)
        self.assertEqual([o['side'] for o in self.submissions],['buy','sell'])
        self.assertEqual(self.qty,0)
        self.assertTrue(self.j.halted.is_set())

    async def test_empty_macro_streams_and_risk_samples_without_buy(self):
        await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,duration=2)
        self.assertEqual(self.submissions,[])
        self.assertGreaterEqual(self.j.db.execute("SELECT count(*) FROM events WHERE kind='RISK_SAMPLE'").fetchone()[0],2)

    async def test_risk_cancels_inflight_strategy_order_before_waiting_for_gate(self):
        self.monthly_signal()
        self.order_behavior = 'working_partial'
        self.trigger_breach = True
        started = time.monotonic()
        await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,duration=5)
        self.assertEqual([o['side'] for o in self.submissions], ['buy','sell'])
        self.assertEqual(self.qty, 0)
        self.assertLess(time.monotonic() - started, 5)

    async def test_check_is_read_only_at_broker(self):
        await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,check=True)
        self.assertEqual(self.submissions,[])
        self.assertFalse(self.j.halted.is_set())

    async def test_kraken_signed_rest_and_native_socket_order(self):
        cfg={**self.cfg,'broker':'kraken','symbol':'BTC/USD'}
        secret=base64.b64encode(b'synthetic-secret').decode()
        async with g.Kraken(cfg,'SYNTHETIC-KEY',secret,self.j) as adapter:
            await adapter.prepare()
            private=asyncio.create_task(adapter.account_stream())
            try:
                await asyncio.wait_for(adapter.balance_ready.wait(),2)
                snap=await adapter.snapshot()
                self.assertEqual(snap['equity'],D('500'))
                cid=self.j.intent('buy',D('.01'),D('100'),'2026-10')
                bid=await adapter.submit(cid,'buy',D('.01'))
                result=await adapter.terminal(bid)
                self.j.receipt(cid,result)
                self.assertEqual(result['fee'],D('.01'))
                self.assertEqual(self.submissions[0]['order_qty'],'0.01')
                bar=await anext(adapter.bars())
                self.assertEqual(bar['price'],D('100'))
            finally:
                private.cancel()
                await asyncio.gather(private,return_exceptions=True)

    async def test_kraken_complete_session_and_shutdown_liquidation(self):
        self.j.close()
        self.root = self.root / 'kraken-project'
        g.initialize(self.root, 'kraken', 'BTC/USD')
        self.cfg = g.configuration(self.root)
        self.j = g.Journal(self.root, self.cfg, 'synthetic')
        await self.start_observer()
        self.monthly_signal()
        secret = base64.b64encode(b'synthetic-secret').decode()
        await g.trading(self.root, self.cfg, 'SYNTHETIC-KEY', secret, self.j, duration=2)
        self.assertEqual([r['side'] for r in self.submissions], ['buy', 'sell'])
        self.assertEqual(self.qty, 0)
        self.assertEqual(self.j.db.execute('SELECT count(*) FROM receipts').fetchone()[0], 2)
        self.assertGreater(self.j.db.execute("SELECT count(*) FROM events WHERE kind='LIQUIDATION_CONFIRMED'").fetchone()[0], 0)
        self.j.verify()

    async def test_unknown_sell_is_never_blindly_duplicated(self):
        self.qty = D('1')
        self.cash = D('400')
        self.j.intent('sell', D('1'), D('100'), emergency=True)
        async with g.Alpaca(self.cfg, 'SYNTHETIC-KEY', 'SYNTHETIC-SECRET', self.j) as adapter:
            await adapter.prepare()
            result = await g.liquidate(adapter, self.j, 'synthetic_unknown_sell')
        self.assertFalse(result)
        self.assertEqual(self.submissions, [])
        self.assertEqual(self.qty, D('1'))

    async def test_external_cash_movement_halts_and_flattens(self):
        with self.j.mutex, self.j.db:
            self.j.set('initialized', True)
            self.j.set('quantity', '0')
        self.cash = D('600')
        worker = g.RiskWorker(self.cfg, 'SYNTHETIC-KEY', 'SYNTHETIC-SECRET', self.j)
        worker.start()
        try:
            deadline = time.monotonic() + 3
            while worker.is_alive() and time.monotonic() < deadline:
                await asyncio.sleep(.05)
            self.assertTrue(self.j.halted.is_set())
            self.assertEqual(self.j.get('halt')['reason'], 'account_reconciliation_mismatch')
        finally:
            worker.stop_event.set()
            while worker.is_alive(): await asyncio.sleep(.05)

    async def test_rejected_or_stale_bar_cannot_order(self):
        self.monthly_signal()
        original = g.Alpaca.bars
        async def stale(adapter):
            async for row in original(adapter):
                row['at'] = '2001-01-01T00:00:00+00:00'
                yield row
        with patch.object(g.Alpaca, 'bars', stale):
            await g.trading(self.root, self.cfg, 'SYNTHETIC-KEY', 'SYNTHETIC-SECRET', self.j, duration=2)
        self.assertEqual(self.submissions, [])
        self.assertTrue(self.j.halted.is_set())

    async def test_unknown_submission_recovered_without_resubmission(self):
        async with g.Alpaca(self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j) as adapter:
            await adapter.prepare()
            cid = self.j.intent('buy', D('1'), D('100'), '2026-10')
            await adapter.submit(cid, 'buy', D('1'))
            # Simulates loss of the response before the local broker-id commit.
            await g.reconcile_pending(adapter, self.j)
        self.assertEqual(len(self.submissions), 1)
        self.assertEqual(self.j.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0], 0)
        self.assertEqual(D(self.j.get('quantity')), D('1'))

    async def test_explicit_rearm_preserves_risk_history(self):
        with self.j.mutex,self.j.db:
            self.j.set('initialized',True)
            self.j.set('quantity','0')
            self.j.set('peak','1000')
            self.j.set('ratchet','500')
        self.j.halt('operator_stop')
        await g.rearm(self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,'synthetic_operator_review')
        self.assertFalse(self.j.halted.is_set())
        self.assertEqual(self.j.get('peak'),'1000')
        self.assertEqual(self.j.get('ratchet'),'500')
        self.assertEqual(self.submissions,[])

    async def test_rearm_refuses_unresolved_intent(self):
        self.j.intent('buy', D('1'), D('100'))
        self.j.halt('test')
        with self.assertRaises(g.Fault):
            await g.rearm(self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,'test')
        self.assertTrue(self.j.halted.is_set())
        self.assertEqual(self.submissions,[])

    async def test_missing_observer_blocks_strategy_without_stopping_risk(self):
        self.monthly_signal()
        self.j.observer=None
        await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,duration=2)
        self.assertEqual(self.submissions,[])
        self.assertEqual(self.j.db.execute("SELECT count(*) FROM events WHERE kind='DECISION'").fetchone()[0],0)
        self.assertGreater(self.j.db.execute("SELECT count(*) FROM events WHERE kind='ENTRY_BLOCKED'").fetchone()[0],0)
        self.assertGreater(self.j.db.execute("SELECT count(*) FROM events WHERE kind='RISK_SAMPLE'").fetchone()[0],0)

    async def test_revoked_gate_never_reaches_decide(self):
        self.monthly_signal()
        journal=self.j
        class ExpiringFixtureGate(FixtureGate):
            def require(self):
                with journal.mutex:
                    if journal.db.execute("SELECT count(*) FROM events WHERE kind='MARKET_TICK'").fetchone()[0]:
                        raise g.AuthenticationFailed()
        self.j.gate=ExpiringFixtureGate()
        await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,duration=2)
        self.assertEqual(self.submissions,[])
        self.assertEqual(self.j.db.execute("SELECT count(*) FROM events WHERE kind='DECISION'").fetchone()[0],0)
        self.assertTrue(self.j.halted.is_set())

    async def test_complete_https_gate_broker_dashboard_and_ledger_path(self):
        from test_auth import GatewayTests
        gateway=GatewayTests(methodName='test_exact_challenge_and_cookie_bound_bootstrap')
        await gateway.asyncSetUp()
        try:
            await gateway.gate.open()
            self.j.gate=gateway.gate
            self.monthly_signal()
            await g.trading(self.root,self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,duration=2)
            self.assertEqual([o['side'] for o in self.submissions],['buy','sell'])
            self.assertEqual(gateway.requests[0][0],'gate')
            self.assertEqual(gateway.requests[1][0],'bootstrap')
            self.assertTrue(list(self.root.glob('live_diagnostics_*.html')))
            self.assertTrue(self.j.verify()['verified'])
            self.assertEqual(self.qty,0)
        finally:
            await gateway.asyncTearDown()

    async def test_authenticated_rebind_changes_only_metadata_and_preserves_halt(self):
        async with g.Alpaca(self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j) as adapter:
            await adapter.prepare();await g.bind_broker_session(adapter,self.j,'SYNTHETIC-KEY','SYNTHETIC-SECRET')
        self.j.halt('synthetic_reviewed_stop')
        old_peak=self.j.get('peak');old_ratchet=self.j.get('ratchet')
        self.j.close()
        self.cfg={**self.cfg,'max_order_usd':'90'}
        self.j=g.Journal(self.root,self.cfg,'synthetic-rotated-key',defer_binding_review=True)
        self.j.gate=FixtureGate();self.j.observer=self.observer
        self.assertTrue(self.j.binding_mismatch)
        await g.approve_local_binding(self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,'synthetic_review')
        self.assertFalse(self.j.binding_mismatch);self.assertTrue(self.j.halted.is_set())
        self.assertEqual(self.j.get('halt')['reason'],'synthetic_reviewed_stop')
        self.assertEqual(self.j.get('peak'),old_peak);self.assertEqual(self.j.get('ratchet'),old_ratchet)
        self.assertEqual(self.submissions,[])

    async def test_rebind_refuses_pending_intent(self):
        self.j.intent('buy',D('1'),D('100'))
        self.j.close()
        self.j=g.Journal(self.root,self.cfg,'synthetic-rotated-key',defer_binding_review=True)
        self.j.gate=FixtureGate();self.j.observer=self.observer
        with self.assertRaises(g.Fault):
            await g.approve_local_binding(self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j,'synthetic_review')
        self.assertTrue(self.j.binding_mismatch);self.assertEqual(self.submissions,[])

    async def test_risk_worker_samples_without_strategy_loop(self):
        # Network server remains on this loop; risk runs in its own OS thread.
        worker=g.RiskWorker(self.cfg,'SYNTHETIC-KEY','SYNTHETIC-SECRET',self.j)
        worker.start()
        try:
            await asyncio.sleep(2.3)
            self.assertTrue(worker.ready.is_set())
            self.assertFalse(worker.failed)
            self.assertGreaterEqual(self.j.db.execute("SELECT count(*) FROM events WHERE kind='RISK_SAMPLE'").fetchone()[0],3)
        finally:
            worker.stop_event.set()
            while worker.is_alive(): await asyncio.sleep(.05)


if __name__=='__main__':
    unittest.main(verbosity=2)