// Automation Runtime — loads scripts, polls market data, dispatches events
// Supports real-time WebSocket feeds with REST polling as fallback heartbeat

import { existsSync, readFileSync, writeFileSync, mkdirSync } from 'fs';
import path from 'path';
import os from 'os';
import { HyperliquidClient } from '../core/client.js';
import {
  roundPrice, roundSize, sleep, normalizeCoin,
  formatUsd, formatPercent, annualizeFundingRate,
} from '../core/utils.js';
import { WebSocketManager } from '../core/ws.js';
import { AutomationRealtimeData } from './realtime.js';
import { AutomationEventBus } from './events.js';
import { loadAutomation } from './loader.js';
import { CLIENT_WRITE_METHODS, createGuardrailedClient } from './guardrails.js';
import { registerAutomation, unregisterAutomation, getRegisteredAutomations as getRegisteredFromFile } from './registry.js';
import { createAutomationAudit, toSerializable, type AutomationAuditSink } from './audit.js';
import { startKeepAwake, type KeepAwakeHandle } from './keep-awake.js';
import type {
  AutomationAPI,
  AutomationAuditObserver,
  AutomationEventPayloads,
  AutomationEventType,
  AutomationLogger,
  AutomationState,
  AutomationSnapshot,
  AutomationAudit,
  PositionSnapshot,
  PublishOptions,
  ScheduledTask,
  RunningAutomation,
  LoadedAutomation,
} from './types.js';

// ── Observer fan-out ────────────────────────────────────────────────
//
// External monitoring packages (e.g. `openbroker-monitoring`) plug into
// the audit pipeline as observers. We auto-load them via convention
// dynamic-import: any package whose default export is a factory returning
// an `AutomationAuditObserver` (or null) and that resolves through Node's
// module resolver gets wired in at startup.

const CONVENTION_OBSERVER_PACKAGES = ['openbroker-monitoring'];

type ObserverFactory =
  | AutomationAuditObserver
  | ((opts?: unknown) => AutomationAuditObserver | null | undefined);

async function loadConventionObservers(log: AutomationLogger): Promise<AutomationAuditObserver[]> {
  const observers: AutomationAuditObserver[] = [];
  for (const name of CONVENTION_OBSERVER_PACKAGES) {
    try {
      const mod = await import(name);
      const exported: ObserverFactory | undefined = mod.default ?? mod;
      const observer = typeof exported === 'function' ? exported() : exported;
      if (observer && typeof observer === 'object') {
        observers.push(observer as AutomationAuditObserver);
        log.info(`[audit-observer] loaded: ${name}`);
      } else {
        // Package resolved but its factory returned null/undefined — typically
        // because required env (e.g. OB_DASHBOARD_URL) is missing. Surface this
        // so operators don't silently lose telemetry.
        log.warn(
          `[audit-observer] ${name} resolved but factory returned no observer — check the package's required env vars`
        );
      }
    } catch (err) {
      const code = (err as NodeJS.ErrnoException | undefined)?.code;
      if (code === 'ERR_MODULE_NOT_FOUND' || code === 'MODULE_NOT_FOUND') {
        log.info(`[audit-observer] ${name} not installed — skipping (install it alongside openbroker to forward telemetry)`);
      } else {
        log.warn(`[audit-observer] failed to load "${name}": ${err instanceof Error ? err.message : String(err)}`);
      }
    }
  }
  return observers;
}

function fanOutNote(observers: AutomationAuditObserver[], kind: string, payload?: unknown): void {
  for (const o of observers) {
    try { o.onNote?.(kind, payload); } catch { /* observer must not break runtime */ }
  }
}

function fanOutMetric(
  observers: AutomationAuditObserver[],
  name: string,
  value: number,
  tags?: Record<string, unknown>,
): void {
  for (const o of observers) {
    try { o.onMetric?.(name, value, tags); } catch { /* observer must not break runtime */ }
  }
}

function fanOutAgentAction(
  observers: AutomationAuditObserver[],
  action: string,
  status: 'success' | 'error',
  details: Record<string, unknown>,
  txHash?: string,
): void {
  for (const o of observers) {
    try { o.onAgentAction?.(action, status, details, txHash); } catch { /* observer must not break runtime */ }
  }
}

const STATE_DIR = path.join(os.homedir(), '.openbroker', 'state');
// ── State persistence ───────────────────────────────────────────────

interface StateController {
  state: AutomationState;
  snapshot(): Record<string, unknown>;
  attachAudit(audit: AutomationAuditSink): void;
}

function createState(id: string): StateController {
  mkdirSync(STATE_DIR, { recursive: true });
  const stateFile = path.join(STATE_DIR, `${id}.json`);

  let data: Record<string, unknown> = {};
  let audit: AutomationAuditSink | null = null;
  if (existsSync(stateFile)) {
    try {
      data = JSON.parse(readFileSync(stateFile, 'utf-8'));
    } catch {
      data = {};
    }
  }

  let flushTimer: ReturnType<typeof setTimeout> | null = null;
  function scheduleFlush() {
    if (flushTimer) return;
    flushTimer = setTimeout(() => {
      flushTimer = null;
      writeFileSync(stateFile, JSON.stringify(data, null, 2));
    }, 500);
  }

  return {
    state: {
      get<T = unknown>(key: string, defaultValue?: T): T | undefined {
        return (key in data ? data[key] : defaultValue) as T | undefined;
      },
      set<T = unknown>(key: string, value: T): void {
        data[key] = value;
        audit?.recordStateChange('set', key, value);
        scheduleFlush();
      },
      delete(key: string): void {
        const previous = key in data ? data[key] : undefined;
        delete data[key];
        audit?.recordStateChange('delete', key, previous);
        scheduleFlush();
      },
      clear(): void {
        data = {};
        audit?.recordStateChange('clear', null);
        scheduleFlush();
      },
    },
    snapshot(): Record<string, unknown> {
      return toSerializable(data);
    },
    attachAudit(nextAudit: AutomationAuditSink): void {
      audit = nextAudit;
    },
  };
}

// ── Logger ──────────────────────────────────────────────────────────

function createLogger(id: string, verbose: boolean, audit?: AutomationAuditSink): AutomationLogger {
  const prefix = `[auto:${id}]`;
  return {
    info: (msg: string) => {
      audit?.recordLog('info', msg);
      console.log(`${prefix} ${msg}`);
    },
    warn: (msg: string) => {
      audit?.recordLog('warn', msg);
      console.log(`${prefix} ⚠ ${msg}`);
    },
    error: (msg: string) => {
      audit?.recordLog('error', msg);
      console.error(`${prefix} ✗ ${msg}`);
    },
    debug: (msg: string) => {
      if (verbose) {
        audit?.recordLog('debug', msg);
        console.log(`${prefix} … ${msg}`);
      }
    },
  };
}

// ── Dry-run client proxy ────────────────────────────────────────────

function createDryClient(client: HyperliquidClient, log: AutomationLogger): HyperliquidClient {
  return new Proxy(client, {
    get(target, prop, receiver) {
      const value = Reflect.get(target, prop, receiver);
      if (typeof prop === 'string' && CLIENT_WRITE_METHODS.has(prop) && typeof value === 'function') {
        return (...args: unknown[]) => {
          log.info(`[DRY] ${prop}(${args.map(a => JSON.stringify(a)).join(', ')})`);
          return Promise.resolve({ status: 'ok', response: { type: 'dry_run' } });
        };
      }
      return value;
    },
  });
}

function createAuditedClient(
  client: HyperliquidClient,
  audit: AutomationAuditSink,
  dryRun: boolean,
  observers: AutomationAuditObserver[],
): HyperliquidClient {
  return new Proxy(client, {
    get(target, prop, receiver) {
      const value = Reflect.get(target, prop, receiver);
      if (typeof prop === 'string' && CLIENT_WRITE_METHODS.has(prop) && typeof value === 'function') {
        return async (...args: unknown[]) => {
          const actionId = `${prop}:${Date.now()}:${Math.random().toString(16).slice(2)}`;
          audit.recordAction({
            actionId,
            phase: 'request',
            method: prop,
            payload: { args },
            dryRun,
          });

          try {
            const result = await value.apply(target, args);
            audit.recordAction({
              actionId,
              phase: 'response',
              method: prop,
              result,
              dryRun,
            });
            fanOutAgentAction(
              observers,
              prop,
              'success',
              { args: toSerializable(args), result: toSerializable(result), dryRun },
            );
            return result;
          } catch (error) {
            audit.recordAction({
              actionId,
              phase: 'error',
              method: prop,
              error,
              dryRun,
            });
            fanOutAgentAction(
              observers,
              prop,
              'error',
              { args: toSerializable(args), error: String(error), dryRun },
            );
            throw error;
          }
        };
      }
      return value;
    },
  });
}

// ── Snapshot building ───────────────────────────────────────────────

async function buildSnapshot(
  client: HyperliquidClient,
): Promise<AutomationSnapshot> {
  // `metaAndAssetCtxs` contains both mostly-static market metadata and live
  // funding/premium values. The client caches it for market lookups, so clear
  // it before each automation poll snapshot or funding_update events freeze at
  // the first fetched value.
  client.invalidateMetaCache();

  const [state, mids, metaCtxs] = await Promise.all([
    client.getUserStateAll(),
    client.getAllMids(),
    client.getMetaAndAssetCtxs(),
  ]);

  const prices = new Map<string, number>();
  for (const [coin, mid] of Object.entries(mids)) {
    prices.set(coin, parseFloat(mid as string));
  }

  const positions = new Map<string, PositionSnapshot>();
  for (const ap of state.assetPositions) {
    const p = ap.position;
    const size = parseFloat(p.szi);
    if (size === 0) continue;
    positions.set(p.coin, {
      coin: p.coin,
      size,
      entryPrice: parseFloat(p.entryPx),
      positionValue: parseFloat(p.positionValue),
      unrealizedPnl: parseFloat(p.unrealizedPnl),
      liquidationPx: p.liquidationPx ? parseFloat(p.liquidationPx) : null,
      leverage: typeof p.leverage === 'object' ? p.leverage.value : parseFloat(String(p.leverage)),
      marginUsed: parseFloat(p.marginUsed),
    });
  }

  const equity = parseFloat(state.marginSummary.accountValue);
  const marginUsed = parseFloat(state.marginSummary.totalMarginUsed);

  // Build funding rates from asset contexts
  const fundingRates = new Map<string, { rate: number; premium: number }>();
  const addFundingRates = (
    universe: Array<{ name?: string }> | undefined,
    assetCtxs: Array<{ funding?: string | number | null; premium?: string | number | null }> | undefined,
  ) => {
    if (!universe || !assetCtxs) return;
    for (let i = 0; i < universe.length; i++) {
      const meta = universe[i];
      const ctx = assetCtxs[i];
      if (ctx && meta?.name) {
        fundingRates.set(meta.name, {
          rate: parseFloat(String(ctx.funding || '0')),
          premium: parseFloat(String(ctx.premium || '0')),
        });
      }
    }
  };

  if (metaCtxs && Array.isArray(metaCtxs)) {
    for (const group of metaCtxs) {
      addFundingRates(group.universe, group.assetCtxs);
    }
  } else if (metaCtxs) {
    addFundingRates(metaCtxs.meta?.universe, metaCtxs.assetCtxs);
  }

  return {
    prices,
    positions,
    openOrderIds: new Set(), // filled by separate call if needed
    equity,
    marginUsed,
    marginUsedPct: equity > 0 ? (marginUsed / equity) * 100 : 0,
    fundingRates,
    timestamp: Date.now(),
  };
}

// ── Publish (webhook) ───────────────────────────────────────────────

function createPublish(
  automationId: string,
  log: AutomationLogger,
  gatewayPort?: number,
  hooksToken?: string,
): (message: string, options?: PublishOptions) => Promise<boolean> {
  return async (message: string, options?: PublishOptions): Promise<boolean> => {
    const token = hooksToken;
    const port = gatewayPort || 18789;

    if (!token) {
      log.debug('publish() skipped — no hooks token configured (pass --hooks-token, set OPENCLAW_HOOKS_TOKEN before invoking the CLI, or configure plugin.hooksToken)');
      return false;
    }

    const body: Record<string, unknown> = {
      message,
      name: options?.name || `ob-auto-${automationId}`,
      wakeMode: options?.wakeMode || 'now',
    };

    if (options?.deliver !== undefined) body.deliver = options.deliver;
    if (options?.channel) body.channel = options.channel;

    try {
      const res = await fetch(`http://127.0.0.1:${port}/hooks/agent`, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': `Bearer ${token}`,
        },
        body: JSON.stringify(body),
      });

      if (!res.ok) {
        log.warn(`publish() failed: HTTP ${res.status} ${res.statusText}`);
        return false;
      }

      log.debug(`publish() delivered to /hooks/agent (${message.length} chars)`);
      return true;
    } catch (err) {
      log.warn(`publish() error: ${err instanceof Error ? err.message : String(err)}`);
      return false;
    }
  };
}

function createAuditedPublish(
  publish: (message: string, options?: PublishOptions) => Promise<boolean>,
  audit: AutomationAuditSink,
): (message: string, options?: PublishOptions) => Promise<boolean> {
  return async (message: string, options?: PublishOptions): Promise<boolean> => {
    try {
      const delivered = await publish(message, options);
      audit.recordPublish(message, options, delivered);
      return delivered;
    } catch (error) {
      audit.recordError('publish', error);
      throw error;
    }
  };
}

// ── Runtime ─────────────────────────────────────────────────────────

export interface RuntimeOptions {
  scriptPath: string;
  id?: string;
  dryRun?: boolean;
  verbose?: boolean;
  pollIntervalMs?: number;
  /** Gateway port for webhook delivery. Falls back to OPENCLAW_GATEWAY_PORT or 18789 */
  gatewayPort?: number;
  /** Hooks token for webhook auth. Falls back to OPENCLAW_HOOKS_TOKEN */
  hooksToken?: string;
  /** Pre-seed state before the factory function runs (e.g. from --set key=value) */
  initialState?: Record<string, unknown>;
  /**
   * Enable WebSocket-first market/account data and events. REST is used for
   * initial static metadata, minute reconciliation, and automatic fallback
   * while the socket is unavailable.
   * @default true
   */
  useWebSocket?: boolean;
  /**
   * Best-effort host idle-sleep inhibition for the lifetime of the automation.
   * Enabled by default for live runs and disabled by default for dry runs.
   */
  keepAwake?: boolean;
}

/** Registry of all running automations */
const registry = new Map<string, RunningAutomation>();

export function getRunningAutomations(): RunningAutomation[] {
  return [...registry.values()];
}

export function getAutomation(id: string): RunningAutomation | undefined {
  return registry.get(id);
}

/** Get all automations from file-based registry (cross-process visibility) */
export { getRegisteredFromFile as getRegisteredAutomations };

export async function startAutomation(options: RuntimeOptions): Promise<RunningAutomation> {
  const {
    scriptPath,
    dryRun = false,
    verbose = false,
    gatewayPort,
    hooksToken,
    initialState,
    useWebSocket = true,
  } = options;

  // This is the disconnected REST fallback cadence. While WebSocket is live,
  // reconciliation is clamped to at least 60 seconds below.
  const pollIntervalMs = options.pollIntervalMs ?? (useWebSocket ? 30_000 : 10_000);

  const id = options.id || path.basename(scriptPath, '.ts');

  if (registry.has(id)) {
    throw new Error(`Automation "${id}" is already running`);
  }

  const stateController = createState(id);

  // Apply --set flags before the factory function runs so CLI overrides win over persisted state.
  if (initialState) {
    for (const [key, value] of Object.entries(initialState)) {
      stateController.state.set(key, value);
    }
  }

  const eventBus = new AutomationEventBus();

  // Each automation owns its client + realtime cache. This prevents one run
  // from replacing or detaching another run's WebSocket provider when several
  // automations share a host process.
  const rawClient = new HyperliquidClient();
  const audit = createAutomationAudit({
    automationId: id,
    scriptPath,
    dryRun,
    verbose,
    pollIntervalMs,
    useWebSocket,
    accountAddress: rawClient.address,
    walletAddress: rawClient.walletAddress,
    isApiWallet: rawClient.isApiWallet,
    initialState,
    persistedState: stateController.snapshot(),
  });
  stateController.attachAudit(audit);

  const log = createLogger(id, verbose, audit);
  const keepAwakeEnabled = options.keepAwake ?? !dryRun;
  let keepAwake: KeepAwakeHandle | null = null;
  if (keepAwakeEnabled) {
    keepAwake = startKeepAwake(`OpenBroker automation ${id} is running`, log);
    if (keepAwake) {
      log.info(`keep-awake enabled via ${keepAwake.backend}.`);
    }
  }
  const observers = await loadConventionObservers(log);
  let loaded: LoadedAutomation;
  try {
    log.info(`Loading automation: ${scriptPath}`);
    loaded = await loadAutomation(scriptPath, { config: stateController.snapshot() });
  } catch (err) {
    const error = err instanceof Error ? err : new Error(String(err));
    audit.recordError('guardrail_validation', error);
    await audit.stop({
      status: 'error',
      stopReason: 'guardrail_validation_error',
      pollCount: 0,
      eventsEmitted: 0,
    });
    keepAwake?.stop();
    throw error;
  }
  audit.recordNote('guardrails', loaded.guardrails);
  const baseClient = dryRun ? createDryClient(rawClient, log) : rawClient;
  const guardedClient = createGuardrailedClient(baseClient, {
    policy: loaded.guardrails,
    rawClient,
    log,
    onViolation: (error, method, args) => {
      audit.recordNote('guardrail_block', {
        code: error.code,
        message: error.message,
        method,
        args: toSerializable(args),
      });
    },
  });
  const client = createAuditedClient(guardedClient, audit, dryRun, observers);

  const startHooks: Array<() => void | Promise<void>> = [];
  const stopHooks: Array<() => void | Promise<void>> = [];
  const errorHooks: Array<(err: Error) => void | Promise<void>> = [];
  const scheduledTasks: ScheduledTask[] = [];

  // Build the API object
  const publish = createAuditedPublish(createPublish(id, log, gatewayPort, hooksToken), audit);
  const auditApi: AutomationAudit = {
    record: (kind: string, payload?: unknown) => {
      audit.recordNote(kind, payload);
      fanOutNote(observers, kind, payload);
    },
    metric: (name: string, value: number, tags?: Record<string, unknown>) => {
      audit.recordMetric(name, value, tags);
      fanOutMetric(observers, name, value, tags);
    },
  };
  const api: AutomationAPI = {
    client,
    utils: { roundPrice, roundSize, sleep, normalizeCoin, formatUsd, formatPercent, annualizeFundingRate },
    on: (event, handler) => eventBus.on(event, handler),
    every: (intervalMs, handler) => scheduledTasks.push({ intervalMs, handler, lastRun: Date.now() }),
    onStart: (handler) => startHooks.push(handler),
    onStop: (handler) => stopHooks.push(handler),
    onError: (handler) => errorHooks.push(handler),
    publish,
    state: stateController.state,
    log,
    audit: auditApi,
    id,
    dryRun,
    guardrails: loaded.guardrails,
  };

  try {
    // Execute the already validated factory function (registers handlers).
    await loaded.factory(api);
  } catch (err) {
    const error = err instanceof Error ? err : new Error(String(err));
    audit.recordError('startup', error);
    await audit.stop({
      status: 'error',
      stopReason: 'startup_error',
      pollCount: 0,
      eventsEmitted: 0,
    });
    keepAwake?.stop();
    throw error;
  }

  // Polling state (declared early so WebSocket handlers can reference)
  let previousSnapshot: AutomationSnapshot | null = null;
  let pollCount = 0;
  let eventsEmitted = 0;
  let isPolling = false;
  let stopped = false;
  let automationReady = false;

  async function handleErrors(errors: Error[]) {
    for (const err of errors) {
      audit.recordError('handler', err);
      log.error(`Handler error: ${err.message}`);
      for (const hook of errorHooks) {
        try { await hook(err); } catch { /* swallow */ }
      }
    }
  }

  function shouldPersistEvent(event: AutomationEventType): boolean {
    return event !== 'tick' && event !== 'price_change';
  }

  async function emitAutomationEvent<E extends AutomationEventType>(
    event: E,
    payload: AutomationEventPayloads[E],
    source: 'poll' | 'ws' | 'manual',
  ): Promise<void> {
    if (shouldPersistEvent(event)) {
      audit.recordEvent(event, source, payload);
    }
    const errors = await eventBus.emit(event, payload);
    if (errors.length) await handleErrors(errors);
    eventsEmitted++;
  }

  // ── WebSocket setup ─────────────────────────────────────────────
  let ws: WebSocketManager | null = null;
  let wsConnected = false;
  let realtimeData: AutomationRealtimeData | null = null;
  // Track latest prices from WebSocket for real-time price_change events
  let wsPrices = new Map<string, number>();
  let wsFundingRates = new Map<string, number>();

  if (useWebSocket) {
    try {
      ws = new WebSocketManager(verbose);
      const unified = await rawClient.isUnifiedAccount().catch(() => null);
      const orderDexes = await rawClient.getPerpDexs()
        .then((dexes) => dexes.slice(1).flatMap((dex) => dex?.name ? [dex.name] : []))
        .catch(() => [] as string[]);
      realtimeData = new AutomationRealtimeData(
        ws,
        rawClient,
        rawClient.address,
        unified,
        orderDexes,
      );
      rawClient.setRealtimeDataProvider(realtimeData);

      // Wire WebSocket events to the automation event bus
      ws.on('allMids', ({ mids }) => {
        const now = Date.now();
        for (const [coin, mid] of Object.entries(mids)) {
          const newPrice = parseFloat(mid);
          if (isNaN(newPrice) || newPrice === 0) continue;
          const oldPrice = wsPrices.get(coin);
          wsPrices.set(coin, newPrice);

          if (automationReady && oldPrice !== undefined && oldPrice !== 0 && eventBus.has('price_change')) {
            const changePct = ((newPrice - oldPrice) / oldPrice) * 100;
            if (Math.abs(changePct) >= 0.01) {
              void emitAutomationEvent('price_change', { coin, oldPrice, newPrice, changePct }, 'ws');
            }
          }
        }
      });

      ws.on('allDexsAssetCtxs', ({ ctxs }) => {
        if (!automationReady || !eventBus.has('funding_update')) return;
        const next = rawClient.fundingRatesFromWs(ctxs);
        for (const [coin, data] of next) {
          const previous = wsFundingRates.get(coin);
          wsFundingRates.set(coin, data.rate);
          if (previous === undefined || previous === data.rate) continue;
          void emitAutomationEvent('funding_update', {
            coin,
            fundingRate: data.rate,
            annualized: annualizeFundingRate(data.rate),
            premium: data.premium,
          }, 'ws');
        }
      });

      ws.on('orderUpdate', (update) => {
        audit.recordOrderUpdate({
          coin: update.order.coin,
          oid: update.order.oid,
          side: update.order.side === 'B' ? 'buy' : 'sell',
          size: parseFloat(update.order.sz),
          price: parseFloat(update.order.limitPx),
          origSize: parseFloat(update.order.origSz),
          status: update.status,
          statusTimestamp: update.statusTimestamp,
          raw: update,
        });

        if (automationReady && eventBus.has('order_update')) {
          void emitAutomationEvent('order_update', {
            coin: update.order.coin,
            oid: update.order.oid,
            side: update.order.side === 'B' ? 'buy' : 'sell',
            size: parseFloat(update.order.sz),
            price: parseFloat(update.order.limitPx),
            origSize: parseFloat(update.order.origSz),
            status: update.status,
            statusTimestamp: update.statusTimestamp,
          }, 'ws');
        }

        // NOTE: order_filled is emitted from the userFill handler below, not from
        // here. The previous implementation fired it from orderUpdate.status
        // === 'filled' using update.order.sz as the size, but that field is the
        // REMAINING size (0 on a terminal fill), not the fill delta — so every
        // consumer saw size=0. Additionally, Hyperliquid does not emit
        // orderUpdate events for pure partial fills that don't transition
        // status, so partial fills were silently dropped entirely. Sourcing
        // order_filled from userFill fixes both issues: sz there IS the fill
        // delta, and the userFills stream fires on every fill (partial and
        // terminal).
      });

      ws.on('userFill', (fill) => {
        audit.recordFill({
          coin: fill.coin,
          side: fill.side === 'B' ? 'buy' : 'sell',
          size: fill.sz,
          price: fill.px,
          time: fill.time,
          closedPnl: fill.closedPnl,
          fee: fill.fee,
          oid: fill.oid,
          crossed: fill.crossed,
        }, fill.time);
        log.debug(`Fill: ${fill.side === 'B' ? 'BUY' : 'SELL'} ${fill.sz} ${fill.coin} @ ${fill.px} (PnL: ${fill.closedPnl})`);

        // Emit order_filled with the authoritative fill delta + fee/pnl from
        // the userFills WS stream. Covers both partial and terminal fills.
        // Fee is converted to USD using feeToken: for non-USDC fees (spot
        // buys pay in the received asset), fee × price yields USD since the
        // fee token is the base of the traded pair and `price` is quote/base.
        if (automationReady && eventBus.has('order_filled')) {
          const size = parseFloat(fill.sz);
          const price = parseFloat(fill.px);
          const rawFee = parseFloat(fill.fee);
          const closedPnl = parseFloat(fill.closedPnl);
          const feeToken = fill.feeToken;
          let feeUsd: number | undefined;
          if (Number.isFinite(rawFee)) {
            if (feeToken === 'USDC' || !feeToken) {
              feeUsd = rawFee;
            } else if (Number.isFinite(price) && price > 0) {
              feeUsd = rawFee * price;
            } else {
              feeUsd = undefined;
            }
          }
          void emitAutomationEvent('order_filled', {
            coin: fill.coin,
            oid: fill.oid,
            side: fill.side === 'B' ? 'buy' : 'sell',
            size,
            price,
            fee: feeUsd,
            feeToken,
            closedPnl: Number.isFinite(closedPnl) ? closedPnl : undefined,
            crossed: fill.crossed,
          }, 'ws');
        }
      });

      ws.on('userEvent', (event) => {
        audit.recordUserEvent(event);
        // Handle liquidation events — only available through WebSocket
        if (automationReady && 'liquidation' in event && eventBus.has('liquidation')) {
          const liq = event.liquidation;
          void emitAutomationEvent('liquidation', {
            lid: liq.lid,
            liquidator: liq.liquidator,
            liquidatedUser: liq.liquidated_user,
            liquidatedNtlPos: parseFloat(liq.liquidated_ntl_pos),
            liquidatedAccountValue: parseFloat(liq.liquidated_account_value),
          }, 'ws');
        }
      });

      ws.on('error', ({ error }) => {
        audit.recordError('websocket', error);
        log.warn(`WebSocket error: ${error.message}`);
      });

      ws.on('disconnected', () => {
        wsConnected = false;
        log.warn('WebSocket disconnected — falling back to REST polling');
      });

      ws.on('connected', () => {
        wsConnected = true;
        log.info('WebSocket connected — real-time events active');
      });

      // Connect subscriptions and warm only the main static universe in
      // parallel. HIP-3 metadata is loaded lazily when referenced.
      const userAddress = rawClient.address as `0x${string}`;
      await Promise.all([
        ws.subscribeAll(userAddress, orderDexes),
        rawClient.initializeRealtimeMetadata().catch((error) => {
          log.warn(`Realtime metadata warmup failed: ${error instanceof Error ? error.message : String(error)}; REST fallback remains available`);
        }),
      ]);
      const seeded = await realtimeData.waitUntilReady();
      const readiness = realtimeData.readinessSummary();
      log.info(
        `WebSocket subscriptions active (mids, asset contexts, account state, spot state, orders, fills, user events)${seeded ? '' : ` · initial snapshots incomplete; REST fallback armed · openOrders ${readiness.seededOrderDexes}/${readiness.expectedOrderDexes}`}`,
      );
    } catch (err) {
      const error = err instanceof Error ? err : new Error(String(err));
      audit.recordError('websocket_setup', error);
      if (verbose && error.stack) {
        log.debug(`WebSocket setup stack: ${error.stack}`);
      }
      log.warn(`WebSocket setup failed: ${error.message} — using REST polling only`);
      rawClient.setRealtimeDataProvider(null);
      realtimeData = null;
      ws = null;
      wsConnected = false;
    }
  }

  async function poll() {
    if (isPolling || stopped) return;
    isPolling = true;

    try {
      const snapshot = await buildSnapshot(rawClient);
      pollCount++;
      const now = Date.now();
      audit.recordSnapshot({
        pollCount,
        equity: snapshot.equity,
        marginUsed: snapshot.marginUsed,
        marginUsedPct: snapshot.marginUsedPct,
        positions: [...snapshot.positions.values()],
        timestamp: now,
      });

      // Always emit tick
      await emitAutomationEvent('tick', { timestamp: now, pollCount }, 'poll');

      if (previousSnapshot) {
        // Price changes (skip when WebSocket is handling real-time prices)
        if (eventBus.has('price_change') && !wsConnected) {
          for (const [coin, newPrice] of snapshot.prices) {
            const oldPrice = previousSnapshot.prices.get(coin);
            if (oldPrice === undefined || oldPrice === 0) continue;
            const changePct = ((newPrice - oldPrice) / oldPrice) * 100;
            if (Math.abs(changePct) >= 0.01) { // 0.01% minimum to fire (filters rounding noise)
              await emitAutomationEvent('price_change', { coin, oldPrice, newPrice, changePct }, 'poll');
            }
          }
        }

        // Funding updates
        if (eventBus.has('funding_update') && !wsConnected) {
          for (const [coin, data] of snapshot.fundingRates) {
            await emitAutomationEvent('funding_update', {
              coin,
              fundingRate: data.rate,
              annualized: annualizeFundingRate(data.rate),
              premium: data.premium,
            }, 'poll');
          }
        }

        // Position opened
        if (eventBus.has('position_opened')) {
          for (const [coin, pos] of snapshot.positions) {
            if (!previousSnapshot.positions.has(coin)) {
              await emitAutomationEvent('position_opened', {
                coin,
                side: pos.size > 0 ? 'long' : 'short',
                size: Math.abs(pos.size),
                entryPrice: pos.entryPrice,
              }, 'poll');
            }
          }
        }

        // Position closed
        if (eventBus.has('position_closed')) {
          for (const [coin, prevPos] of previousSnapshot.positions) {
            if (!snapshot.positions.has(coin)) {
              await emitAutomationEvent('position_closed', {
                coin,
                previousSize: prevPos.size,
                entryPrice: prevPos.entryPrice,
              }, 'poll');
            }
          }
        }

        // Position size changed
        if (eventBus.has('position_changed')) {
          for (const [coin, pos] of snapshot.positions) {
            const prevPos = previousSnapshot.positions.get(coin);
            if (prevPos && pos.size !== prevPos.size) {
              await emitAutomationEvent('position_changed', {
                coin,
                oldSize: prevPos.size,
                newSize: pos.size,
                entryPrice: pos.entryPrice,
              }, 'poll');
            }
          }
        }

        // PnL threshold (5% of position value)
        if (eventBus.has('pnl_threshold')) {
          for (const [coin, pos] of snapshot.positions) {
            const prevPos = previousSnapshot.positions.get(coin);
            if (!prevPos || pos.positionValue === 0) continue;
            const pnlChange = Math.abs(pos.unrealizedPnl - prevPos.unrealizedPnl);
            const changePct = (pnlChange / pos.positionValue) * 100;
            if (changePct >= 5) {
              await emitAutomationEvent('pnl_threshold', {
                coin,
                unrealizedPnl: pos.unrealizedPnl,
                changePct,
                positionValue: pos.positionValue,
              }, 'poll');
            }
          }
        }

        // Margin warning (80%)
        if (eventBus.has('margin_warning') && snapshot.marginUsedPct >= 80) {
          const prevPct = previousSnapshot.marginUsedPct;
          if (prevPct < 80 || snapshot.marginUsedPct - prevPct >= 5) {
            await emitAutomationEvent('margin_warning', {
              marginUsedPct: snapshot.marginUsedPct,
              equity: snapshot.equity,
              marginUsed: snapshot.marginUsed,
            }, 'poll');
          }
        }

        // Order filled — compare open order IDs
        // (Skipped for MVP — requires tracking open orders per poll, will add when needed)
      }

      previousSnapshot = snapshot;
    } catch (err) {
      const error = err instanceof Error ? err : new Error(String(err));
      audit.recordError('poll', error);
      log.error(`Poll error: ${error.message}`);
    } finally {
      isPolling = false;
    }
  }

  async function runScheduledTasks(): Promise<void> {
    const now = Date.now();
    for (const task of scheduledTasks) {
      if (task.running || now - task.lastRun < task.intervalMs) continue;
      task.running = true;
      task.lastRun = now;
      try {
        await task.handler();
      } catch (err) {
        const error = err instanceof Error ? err : new Error(String(err));
        audit.recordError('scheduled_task', error);
        log.error(`Scheduled task error: ${error.message}`);
        await handleErrors([error]);
      } finally {
        task.running = false;
      }
    }
  }

  // Seed an audit snapshot. With a healthy socket this is assembled from the
  // live cache; REST is used only for any feed that has not produced its first
  // snapshot yet.
  await poll();

  // Start hooks run after WebSocket caches are seeded, so strategy reads are
  // WebSocket-first from their very first decision.
  for (const hook of startHooks) {
    try {
      await hook();
    } catch (err) {
      const error = err instanceof Error ? err : new Error(String(err));
      audit.recordError('onStart', error);
      log.error(`onStart hook error: ${error.message}`);
      for (const errorHook of errorHooks) {
        try { await errorHook(error); } catch { /* swallow */ }
      }
    }
  }
  automationReady = true;

  // api.every() is a real scheduler now; it no longer depends on how often a
  // heavyweight REST reconciliation snapshot runs.
  const scheduleTimer = setInterval(() => { void runScheduledTasks(); }, 500);

  // --poll controls the REST fallback cadence while the socket is down. While
  // connected, reconcile at most once per minute regardless of a shorter
  // fallback interval supplied by older launch commands.
  const restReconcileMs = Math.max(60_000, pollIntervalMs);
  let lastPollAt = Date.now();
  const pollTimer = setInterval(() => {
    const interval = wsConnected ? restReconcileMs : pollIntervalMs;
    if (Date.now() - lastPollAt < interval) return;
    lastPollAt = Date.now();
    void poll();
  }, Math.min(1_000, pollIntervalMs));

  const wsLabel = wsConnected ? ', ws=on' : (useWebSocket ? ', ws=failed' : '');
  log.info(
    `Started (REST fallback ${pollIntervalMs / 1000}s, connected reconcile ${restReconcileMs / 1000}s, dry=${dryRun}${wsLabel})`,
  );

  // Stop function
  async function stop(opts?: { persist?: boolean }) {
    if (stopped) return;
    stopped = true;
    clearInterval(scheduleTimer);
    clearInterval(pollTimer);

    // Close WebSocket
    if (ws) {
      ws.removeAllListeners();
      await ws.close();
      ws = null;
    }
    rawClient.setRealtimeDataProvider(null);
    realtimeData = null;

    for (const hook of stopHooks) {
      try { await hook(); } catch (err) {
        const error = err instanceof Error ? err : new Error(String(err));
        audit.recordError('onStop', error);
        log.error(`onStop hook error: ${error.message}`);
      }
    }

    keepAwake?.stop();

    eventBus.removeAll();
    registry.delete(id);

    // persist defaults to true — fully remove from file registry.
    // When false (gateway shutdown), keep the entry so it restarts next time.
    if (opts?.persist !== false) {
      unregisterAutomation(id);
    }
    log.info(`Stopped (${pollCount} polls, ${eventsEmitted} events)`);
    await audit.stop({
      status: 'stopped',
      stopReason: opts?.persist === false ? 'shutdown_keep_registry' : 'manual_stop',
      pollCount,
      eventsEmitted,
    });
  }

  const entry: RunningAutomation = {
    id,
    scriptPath,
    startedAt: new Date(),
    get pollCount() { return pollCount; },
    get eventsEmitted() { return eventsEmitted; },
    dryRun,
    stop,
  };

  registry.set(id, entry);

  // Persist to file-based registry so other processes (CLI, plugin) can see it
  registerAutomation({
    id,
    scriptPath,
    dryRun,
    verbose,
    pollIntervalMs,
  });

  return entry;
}