"""replay_injector.py — 3-month BTC replay into trade_lbm_v1 with
continuous Fractonaut observation.

Builds on validate_trade_v1.py replay loop. Differences:
  * Source: all of March/April/May 2026 BTC parquets (chronological).
  * No artificial sleep between minutes beyond MINUTE_DT (default 0.10s);
    daemon runs as fast as it can take ZMQ commands.
  * Fractonaut queries are RATE-LIMITED and ASYNC (background worker
    thread) so a slow LLM call does not stall the replay loop.
  * Trigger rules:
      a) Any structural event (consolidation/breakout/support/resistance),
         throttled to max 1 query per 30 replay-minutes; if multiple
         events fire inside the window we pick the highest-magnitude one.
      b) regime_product crossings of 1.0 (quiet->active) and 8.0
         (active->spike) in either direction.
      c) Start of each calendar month (March 1, April 1, May 1, UTC).
      d) End of replay: synthesis query.
  * Per-minute parquet row: minute, mid_price, regime_product, asy, coh,
    last_trade_age_s, events_fired (JSON), fractonaut_queried (bool),
    query_reason (str).
  * Fractonaut chronicle: JSONL tagged with minute, mid_price,
    regime_product, query_reason.
  * Post-run correlation: forward returns at h=60 and h=240 by event
    type / query_reason, written as a Markdown table.

Owner: RESONANCE (tagged via --agent-owner=RESONANCE marker on
command line).
"""
# --agent-owner=RESONANCE
import argparse
import gc
import json
import os
import queue
import signal
import sys
import threading
import time
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import zmq

# ============================== config ===================================
DAEMON_CMD   = "tcp://127.0.0.1:5567"
DAEMON_TELEM = "tcp://127.0.0.1:5566"
FRACT_URL    = "http://127.0.0.1:28822/ask"
FRACT_TO_S   = 240

DATA_ROOT = Path("/mnt/d/PaperTrader/research/hl_data/minutes")
RUN_ROOT  = Path("/mnt/d/Resonance_Engine/traj")

COIN          = "BTC"
NX            = 512
TICK_USD      = 1.0
BOOK_DECAY_L  = 30
MINUTE_DT     = 0.10            # wall-seconds per replayed minute
RELAX_BOOT_S  = 4.0             # let field absorb first book before replay
WARMUP_MIN    = 30              # minutes ignored before arm_consolidation
ASK_EVENT_GAP = 30              # min replay-minutes between event queries
ASK_GLOBAL_GAP = 60             # min replay-minutes between ANY non-monthly query
                                # (sized so Ollama @ ~4s/query keeps up at 9 min/s replay)
ROLL_FLUSH_N  = 2000            # flush parquet every N minutes

# regime_product crossing thresholds (level low, level high)
RP_THRESH = [1.0, 8.0]
# Hysteresis: regime category must persist this many minutes before a
# state-change query fires. Stops oscillation around 1.0 from triggering
# on every flap.
RP_HYSTERESIS_MIN = 3

running = True


def _sig(sig, _f):
    global running
    print(f"\n[REPLAY] signal {sig} — graceful shutdown")
    sys.stdout.flush()
    running = False


signal.signal(signal.SIGINT, _sig)
signal.signal(signal.SIGTERM, _sig)


# ============================== ZMQ ======================================
def make_pub(ctx, ep):
    s = ctx.socket(zmq.PUB)
    s.setsockopt(zmq.SNDHWM, 4096)
    s.connect(ep)
    return s


def make_sub(ctx, ep, topic=b""):
    s = ctx.socket(zmq.SUB)
    s.setsockopt(zmq.SUBSCRIBE, topic)
    s.setsockopt(zmq.RCVHWM, 8192)
    s.connect(ep)
    return s


def send_cmd(pub, obj):
    pub.send_string(json.dumps(obj))


def drain_telem(sub, max_ms=80):
    out = []
    poller = zmq.Poller(); poller.register(sub, zmq.POLLIN)
    deadline = time.time() + max_ms / 1000.0
    while time.time() < deadline:
        socks = dict(poller.poll(timeout=10))
        if sub in socks and socks[sub] == zmq.POLLIN:
            try:
                raw = sub.recv_string(zmq.NOBLOCK)
                try: out.append(json.loads(raw))
                except Exception: pass
            except zmq.Again:
                pass
        else:
            break
    return out


# ============================== book =====================================
def synthesize_book(taker_buy_usd: float, taker_sell_usd: float):
    bid = np.zeros(NX, dtype=np.float64)
    ask = np.zeros(NX, dtype=np.float64)
    half = NX // 2
    offs = np.arange(1, half + 1)
    decay = np.exp(-offs / BOOK_DECAY_L)
    decay /= decay.sum()
    bid_vals = max(taker_buy_usd, 0.0) * decay
    ask_vals = max(taker_sell_usd, 0.0) * decay
    bid[half-1::-1] = bid_vals
    ask[half:half+len(decay)] = ask_vals
    total = bid + ask
    m = total.mean()
    if m > 1e-12:
        bid = bid / m
        ask = ask / m
    return bid.tolist(), ask.tolist()


# ============================== Fractonaut worker ========================
class FractWorker:
    """Background thread that drains an ask-queue and writes the chronicle.

    Replay loop calls .enqueue(reason, minute, context_dict) — non-blocking;
    queue is bounded so a stuck Ollama can never starve the replay.
    """

    def __init__(self, chronicle_path: Path, log):
        self.q       = queue.Queue(maxsize=32)
        self.chron   = chronicle_path
        self.log     = log
        self.thread  = threading.Thread(target=self._run, daemon=True)
        self.n_ok    = 0
        self.n_err   = 0
        self.n_drop  = 0
        self.stop    = False
        self.thread.start()

    def enqueue(self, reason: str, minute: int, ctx: dict, question: str):
        item = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "reason": reason, "minute": minute,
            "ctx": ctx, "question": question,
        }
        try:
            self.q.put_nowait(item)
        except queue.Full:
            self.n_drop += 1
            self.log(f"[FW] drop (queue full) reason={reason} min={minute}")

    def _ask(self, question: str):
        body = json.dumps({"question": question}).encode()
        req  = urllib.request.Request(
            FRACT_URL, data=body, headers={"Content-Type": "application/json"})
        t0 = time.time()
        try:
            with urllib.request.urlopen(req, timeout=FRACT_TO_S) as r:
                data = json.loads(r.read())
            return {"ok": True, "elapsed_s": round(time.time()-t0, 2),
                    "response": data.get("response", ""),
                    "turn": data.get("turn"), "model": data.get("model")}
        except Exception as e:
            return {"ok": False, "elapsed_s": round(time.time()-t0, 2),
                    "error": str(e)}

    def _run(self):
        while not self.stop or not self.q.empty():
            try:
                item = self.q.get(timeout=0.5)
            except queue.Empty:
                continue
            res = self._ask(item["question"])
            entry = {**item, "result": res}
            try:
                with open(self.chron, "a", encoding="utf-8") as f:
                    f.write(json.dumps(entry) + "\n")
            except Exception as e:
                self.log(f"[FW] chronicle write error: {e}")
            if res["ok"]:
                self.n_ok += 1
                self.log(f"[FW] ok reason={item['reason']} min={item['minute']} "
                         f"{res['elapsed_s']}s reply={res['response'][:80]!r}")
            else:
                self.n_err += 1
                self.log(f"[FW] err reason={item['reason']} min={item['minute']} "
                         f"err={res.get('error')}")

    def shutdown(self, drain_timeout_s=600):
        self.stop = True
        t0 = time.time()
        while (not self.q.empty()) and (time.time() - t0 < drain_timeout_s):
            time.sleep(2)
        self.thread.join(timeout=10)


# ============================== data load ================================
def load_three_months(root: Path, coin: str, log) -> pd.DataFrame:
    """Walk root/YYYYMMDD/H.parquet for YYYYMM in {202603,202604,202605}.

    Concatenate, filter to coin, sort by minute, dedupe.
    """
    frames = []
    months = ("202603", "202604", "202605")
    day_dirs = []
    for m in months:
        day_dirs.extend(sorted(d for d in root.iterdir() if d.name.startswith(m)))
    log(f"[DATA] {len(day_dirs)} day-dirs across {months}")
    for i, dd in enumerate(day_dirs):
        for hp in sorted(dd.iterdir()):
            if not hp.name.endswith(".parquet"):
                continue
            try:
                df = pd.read_parquet(hp, columns=[
                    "minute", "coin", "mid_price", "signed_flow_usd",
                    "taker_buy_usd", "taker_sell_usd", "trade_count"
                ])
            except Exception as e:
                log(f"[DATA] skip {hp}: {e}")
                continue
            frames.append(df[df["coin"] == coin])
        if (i + 1) % 10 == 0:
            log(f"[DATA] loaded {i+1}/{len(day_dirs)} days")
    if not frames:
        raise RuntimeError("no data loaded")
    df = pd.concat(frames, ignore_index=True)
    df = df.drop_duplicates(subset=["minute"]).sort_values("minute").reset_index(drop=True)
    df["ts_utc"] = pd.to_datetime(df["minute"] * 60, unit="s", utc=True)
    log(f"[DATA] {len(df):,} BTC minute rows  range=[{df['ts_utc'].iloc[0]} .. {df['ts_utc'].iloc[-1]}]")
    return df


# ============================== correlation ==============================
def compute_correlations(events_parquet: Path, chronicle_path: Path,
                         out_md: Path, df_source: pd.DataFrame, log):
    """For each Fractonaut-queried minute, look up fwd log-returns at
    h=60 and h=240 from the source minute->mid_price index and aggregate
    by reason. Write a Markdown table."""
    ev = pd.read_parquet(events_parquet)
    ev = ev[ev["fractonaut_queried"]].copy()
    src = df_source.set_index("minute")["mid_price"].astype(float)

    def fwd(min_t, h):
        try:
            p0 = float(src.loc[min_t])
            p1 = float(src.loc[min_t + h])
            return float(np.log(p1 / p0))
        except KeyError:
            return np.nan

    ev["fwd_60"]  = ev["minute"].apply(lambda m: fwd(m, 60))
    ev["fwd_240"] = ev["minute"].apply(lambda m: fwd(m, 240))

    # For event-fire rows, drill into events_fired to extract the kind.
    def first_kind(events_json):
        try:
            arr = json.loads(events_json) if isinstance(events_json, str) else (events_json or [])
            if arr: return arr[0].get("kind", "?")
        except Exception:
            pass
        return "?"

    ev["event_kind"] = ev["events_fired"].apply(first_kind)
    ev["bucket"] = np.where(
        ev["query_reason"] == "event",
        "event:" + ev["event_kind"],
        ev["query_reason"]
    )

    grp = ev.groupby("bucket").agg(
        n=("minute", "count"),
        mean_fwd_60=("fwd_60",  "mean"),
        med_fwd_60=("fwd_60",   "median"),
        mean_fwd_240=("fwd_240","mean"),
        med_fwd_240=("fwd_240", "median"),
    ).sort_values("n", ascending=False)

    log("[CORR] bucket summary:")
    log(grp.to_string())

    with open(out_md, "w") as f:
        f.write("# Replay correlation table\n\n")
        f.write(f"generated {datetime.now(timezone.utc).isoformat()}\n\n")
        f.write(f"queried minutes: n={len(ev)}\n\n")
        f.write("| bucket | n | mean_fwd_60 | med_fwd_60 | mean_fwd_240 | med_fwd_240 |\n")
        f.write("|---|---:|---:|---:|---:|---:|\n")
        for b, row in grp.iterrows():
            f.write(f"| {b} | {int(row['n'])} | {row['mean_fwd_60']:+.6f} | "
                    f"{row['med_fwd_60']:+.6f} | {row['mean_fwd_240']:+.6f} | "
                    f"{row['med_fwd_240']:+.6f} |\n")
    log(f"[CORR] wrote {out_md}")


# ============================== main =====================================
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--minute-dt", type=float, default=MINUTE_DT,
                    help="wall seconds per replayed minute (lower = faster)")
    ap.add_argument("--limit", type=int, default=0,
                    help="only replay first N minutes (0=all)")
    ap.add_argument("--run-id", type=str, default=None,
                    help="override run id; default = UTC timestamp")
    ap.add_argument("--dry-run", action="store_true",
                    help="load data + open sockets but do not send commands")
    ap.add_argument("--skip-correlation", action="store_true")
    ap.add_argument("--flow-drive-tc", action="store_true",
                    help="Test C: modulate set_flow_drive by trade_count z-score")
    ap.add_argument("--virtual-recentre", type=float, default=0.0,
                    help="Virtual recentre: only send set_mid when |mid - last_sent_mid| > this USD threshold. 0 = baseline (every minute).")
    ap.add_argument("--no-book", action="store_true",
                    help="Diagnostic: do NOT send set_book. Inject_trade only.")
    ap.add_argument("--cs2", action="store_true",
                    help="Test C: per-minute set_temperature_profile, cold (cs2 down) where trade_count is high. "
                         "Uses tc_z (auto-enables tc_z computation).")
    ap.add_argument("--cs2-alpha", type=float, default=0.30,
                    help="Strength of cold drive: cs2[col] = CS2_NOMINAL * (1 - alpha * clamp(tc_z,0,1)) inside the patch.")
    ap.add_argument("--cs2-floor", type=float, default=0.16,
                    help="Hard host-side floor on cs2 to stay clear of kernel CS2_MIN=0.15.")
    ap.add_argument("--cs2-half-width", type=int, default=50,
                    help="Cold patch half-width in cols around col 256 (mid).")
    ap.add_argument("--window-start-min", type=int, default=0,
                    help="Slice df to start at this minute (epoch//60). 0 = no slice.")
    ap.add_argument("--window-len", type=int, default=0,
                    help="After window-start-min slice, keep only this many minutes. 0 = keep all.")
    args = ap.parse_args()

    run_id = args.run_id or datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    run_dir = RUN_ROOT / f"replay_3month_{run_id}"
    run_dir.mkdir(parents=True, exist_ok=True)
    progress_log = run_dir / "progress.log"

    def log(msg):
        line = f"[{datetime.now(timezone.utc).strftime('%H:%M:%S')}] {msg}"
        print(line, flush=True)
        try:
            with open(progress_log, "a") as f:
                f.write(line + "\n")
        except Exception:
            pass

    log(f"replay_injector starting; run_id={run_id}")
    log(f"run_dir={run_dir}")
    log(f"minute_dt={args.minute_dt}  limit={args.limit}  dry_run={args.dry_run}")

    df = load_three_months(DATA_ROOT, COIN, log)
    if args.window_start_min > 0:
        before = len(df)
        df = df[df["minute"] >= args.window_start_min].reset_index(drop=True)
        log(f"[DATA] sliced from minute>={args.window_start_min}: {before:,} -> {len(df):,}")
    if args.window_len > 0:
        df = df.head(args.window_len).reset_index(drop=True)
        log(f"[DATA] window-len truncated to first {len(df):,} minutes")
    if args.limit > 0:
        df = df.head(args.limit).copy()
        log(f"[DATA] truncated to first {len(df):,} minutes")

    # Precompute rolling trade_count z-score for --flow-drive-tc and/or --cs2
    if args.flow_drive_tc or args.cs2:
        tc = df["trade_count"].astype(float).fillna(0.0)
        roll_mean = tc.rolling(window=240, min_periods=30).mean()
        roll_std  = tc.rolling(window=240, min_periods=30).std().replace(0.0, np.nan)
        df["tc_z"] = ((tc - roll_mean) / roll_std).fillna(0.0).clip(-3.0, 5.0)
        log(f"[CTRL] tc_z computed (flow_drive_tc={args.flow_drive_tc}, cs2={args.cs2})  "
            f"mean={df['tc_z'].mean():.3f} std={df['tc_z'].std():.3f} "
            f"min={df['tc_z'].min():.3f} max={df['tc_z'].max():.3f}")
    if args.cs2:
        log(f"[CTRL] cs2 ENABLED  alpha={args.cs2_alpha:.2f}  floor={args.cs2_floor:.3f}  "
            f"half_width={args.cs2_half_width} cols (patch cols {NX//2 - args.cs2_half_width}..{NX//2 + args.cs2_half_width})")
        CS2_NOMINAL_PY = 1.0 / 3.0
        log(f"[CTRL] cs2 max-cold value (tc_z=1) = {CS2_NOMINAL_PY * (1 - args.cs2_alpha):.4f}  "
            f"(stability bound u_max < sqrt(cs2) = {(CS2_NOMINAL_PY * (1 - args.cs2_alpha))**0.5:.4f})")
    if args.virtual_recentre > 0:
        log(f"[CTRL] virtual-recentre ENABLED  threshold=${args.virtual_recentre:.0f}")

    chronicle_path = run_dir / "fractonaut_replay_chronicle.jsonl"
    events_parquet = run_dir / "replay_events.parquet"
    correl_md      = run_dir / "correlations.md"

    ctx = zmq.Context.instance()
    pub = make_pub(ctx, DAEMON_CMD)
    sub = make_sub(ctx, DAEMON_TELEM)
    time.sleep(0.5)  # slow joiner

    if not args.dry_run:
        log("[CTRL] reset_equilibrium")
        send_cmd(pub, {"cmd": "reset_equilibrium"})
        time.sleep(0.5)
        first = df.iloc[0]
        log(f"[CTRL] set_mid={first['mid_price']:.2f}  set_tick_size={TICK_USD}")
        send_cmd(pub, {"cmd": "set_tick_size", "value": TICK_USD})
        send_cmd(pub, {"cmd": "set_mid", "price": float(first["mid_price"])})
        bid, ask = synthesize_book(float(first["taker_buy_usd"]),
                                   float(first["taker_sell_usd"]))
        if not args.no_book:
            send_cmd(pub, {"cmd": "set_book", "bid": bid, "ask": ask})
        time.sleep(0.3)
        log(f"[CTRL] booting; relaxing {RELAX_BOOT_S}s")
        time.sleep(RELAX_BOOT_S)
        drain_telem(sub, 400)

    fw = FractWorker(chronicle_path, log)
    fw.enqueue("replay_start", int(df.iloc[0]["minute"]), {
        "rows": len(df),
        "start_ts": str(df.iloc[0]["ts_utc"]),
        "end_ts":   str(df.iloc[-1]["ts_utc"]),
    },
    "Three-month BTC replay is starting. Briefly describe the field's "
    "current state before any replay injection begins. This is the "
    "pre-replay baseline.")

    # ─── replay loop ───
    rows_out      = []
    last_event_query_min = -999999
    last_any_query_min   = -999999      # global gate (non-monthly)
    last_rp_band         = None         # 'quiet'/'normal'/'elevated'/'spike'
    band_streak          = 0            # consecutive minutes in current band
    pending_band_change  = None         # (new_band, prev_band, rp_now) waiting on hysteresis
    last_month    = None
    armed         = False
    seen_event_keys = set()
    pending_event = None     # (mag, ev_dict, frame_dict) within the 30-min window
    pending_event_min = None
    minute_count  = 0
    t_start       = time.time()
    total_recentres = 0

    def rp_band(rp):
        if rp is None or not np.isfinite(rp): return None
        if rp < 0.05:  return "quiet"
        if rp < 1.0:   return "normal"
        if rp < 8.0:   return "elevated"
        return "spike"

    log("[LOOP] entering replay")
    last_sent_mid = float(df.iloc[0]["mid_price"]) if len(df) else 0.0
    n_virtual_skipped = 0
    for idx, row in df.iterrows():
        if not running:
            log("[LOOP] interrupted")
            break
        mid_raw = row["mid_price"]
        mid = float(mid_raw) if pd.notna(mid_raw) else np.nan
        if not np.isfinite(mid) or mid <= 0:
            rows_out.append({
                "minute": int(row["minute"]), "mid_price": None,
                "regime_product": None, "asymmetry": None, "coherence": None,
                "last_trade_age_s": None, "events_fired": json.dumps([]),
                "fractonaut_queried": False, "query_reason": "skip_no_mid",
            })
            minute_count += 1
            continue
        tbuy = float(row["taker_buy_usd"] or 0.0)
        tsel = float(row["taker_sell_usd"] or 0.0)
        bid, ask = synthesize_book(tbuy, tsel)

        if not args.dry_run:
            if not armed and minute_count >= WARMUP_MIN:
                send_cmd(pub, {"cmd": "arm_consolidation"})
                armed = True
                log(f"  [t={minute_count}] arm_consolidation")
            # Virtual recentre: skip set_mid unless drift exceeds threshold
            if args.virtual_recentre > 0:
                if abs(mid - last_sent_mid) > args.virtual_recentre:
                    send_cmd(pub, {"cmd": "set_mid", "price": mid})
                    last_sent_mid = mid
                else:
                    n_virtual_skipped += 1
            else:
                send_cmd(pub, {"cmd": "set_mid",  "price": mid})
            if not args.no_book:
                send_cmd(pub, {"cmd": "set_book", "bid": bid, "ask": ask})
            # Test C: modulate background flow drive by trade_count z-score
            if args.flow_drive_tc:
                z = float(row.get("tc_z", 0.0))
                # map z in [-3, +5] to flow_drive in [0, 0.1]; baseline 0.02
                fd = float(np.clip(0.02 + 0.015 * z, 0.0, 0.1))
                send_cmd(pub, {"cmd": "set_flow_drive", "value": fd})
            # Test C (geometric): per-minute lattice temperature.
            # Cold patch around mid (col NX/2 +/- half_width) with strength
            # set by tc_z clipped to [0,1]. Hard host-side floor keeps us
            # above kernel CS2_MIN=0.15 so the daemon never clamps and we
            # don't sit on the stability bound.
            if args.cs2:
                z01 = float(np.clip(row.get("tc_z", 0.0), 0.0, 1.0))
                CS2_NOM = 1.0 / 3.0
                cold = max(CS2_NOM * (1.0 - args.cs2_alpha * z01), args.cs2_floor)
                profile = [CS2_NOM] * NX
                centre = NX // 2
                lo = max(0, centre - args.cs2_half_width)
                hi = min(NX, centre + args.cs2_half_width + 1)
                for col in range(lo, hi):
                    profile[col] = cold
                send_cmd(pub, {"cmd": "set_temperature_profile", "values": profile})
            if tbuy > 0:
                send_cmd(pub, {"cmd": "inject_trade", "price": mid + TICK_USD,
                               "size": tbuy / mid / 100.0,
                               "side": "buy",  "aggressor": True})
            if tsel > 0:
                send_cmd(pub, {"cmd": "inject_trade", "price": mid - TICK_USD,
                               "size": tsel / mid / 100.0,
                               "side": "sell", "aggressor": True})

        time.sleep(args.minute_dt)
        frames = drain_telem(sub, int(args.minute_dt * 1000 * 0.8) + 30)

        # latest scalar state for the row + Fract context
        latest = frames[-1] if frames else {}
        rp     = latest.get("regime_product")
        asy    = latest.get("asymmetry")
        coh    = latest.get("coherence")
        age    = latest.get("last_trade_age_s")
        # Per-minute vel_max (peak across frames seen this minute) for cs² stability watch
        vel_max_min = 0.0
        for fr in frames:
            v = fr.get("vel_max")
            if v is not None and float(v) > vel_max_min:
                vel_max_min = float(v)

        # gather events fired this minute (deduped against history)
        evs_fired = []
        for fr in frames:
            if fr.get("recenter_event"):
                total_recentres += 1
            for ev in (fr.get("events") or []):
                key = (ev.get("kind"), ev.get("col"),
                       round(float(ev.get("price", 0.0)), 2))
                if key in seen_event_keys:
                    continue
                seen_event_keys.add(key)
                evs_fired.append({**ev, "cycle": fr.get("cycle")})

        # ─── trigger decisions ───
        queried = False
        reason  = ""

        # (a) structural events, rate-limited (30 replay-min gap)
        if evs_fired and minute_count >= WARMUP_MIN:
            top = max(evs_fired, key=lambda e: abs(float(e.get("mag", 0.0))))
            mag = abs(float(top.get("mag", 0.0)))
            if pending_event is None or mag > pending_event[0]:
                pending_event = (mag, top, dict(latest))
                pending_event_min = minute_count
        if (pending_event is not None
                and (minute_count - last_event_query_min) >= ASK_EVENT_GAP
                and (minute_count - last_any_query_min) >= ASK_GLOBAL_GAP):
            mag, ev, fctx = pending_event
            q = (f"Structural event: {ev.get('kind')} at column {ev.get('col')} "
                 f"price ${float(ev.get('price', 0.0)):,.2f} magnitude {float(ev.get('mag', 0.0)):.3f}. "
                 f"regime_product={fctx.get('regime_product')}. "
                 f"last_trade_age_s={fctx.get('last_trade_age_s')}. "
                 f"Describe what the spatial structure shows at this moment. "
                 f"Is this event consistent with the surrounding density and "
                 f"divergence profile?")
            fw.enqueue("event", int(row["minute"]), {
                "kind": ev.get("kind"), "col": ev.get("col"),
                "price": ev.get("price"), "mag": ev.get("mag"),
                "regime_product": fctx.get("regime_product"),
                "asymmetry": fctx.get("asymmetry"),
                "coherence":  fctx.get("coherence"),
                "last_trade_age_s": fctx.get("last_trade_age_s"),
                "mid_price": mid,
            }, q)
            queried = True
            reason  = "event"
            last_event_query_min = minute_count
            last_any_query_min   = minute_count
            pending_event = None

        # (b) regime BAND change, hysteresis-gated. Only fire when the
        # band has held for >= RP_HYSTERESIS_MIN consecutive minutes AND
        # the global gate has elapsed. Stops oscillation from flooding.
        cur_band = rp_band(rp)
        if cur_band is not None:
            if cur_band == last_rp_band:
                band_streak += 1
            else:
                if last_rp_band is not None:
                    pending_band_change = (cur_band, last_rp_band, rp)
                band_streak = 1
                last_rp_band = cur_band
            if (not queried and pending_band_change is not None
                    and band_streak >= RP_HYSTERESIS_MIN
                    and (minute_count - last_any_query_min) >= ASK_GLOBAL_GAP):
                new_b, prev_b, rp_now = pending_band_change
                # only ask about transitions across a meaningful threshold
                rank = {"quiet":0, "normal":1, "elevated":2, "spike":3}
                if abs(rank.get(new_b,0) - rank.get(prev_b,0)) >= 1:
                    q = (f"regime_product has settled into the {new_b} band "
                         f"(now {rp:.4f}, previously {prev_b}). What changed in the field?")
                    fw.enqueue("rp_band", int(row["minute"]), {
                        "new_band": new_b, "prev_band": prev_b,
                        "rp_now": rp, "mid_price": mid,
                    }, q)
                    queried = True
                    reason  = f"rp_band_{prev_b}_to_{new_b}"
                    last_any_query_min = minute_count
                pending_band_change = None

        # (c) start of each month (UTC) — always queries, no gate
        cur_month = pd.Timestamp(row["ts_utc"]).month
        if not queried and last_month is not None and cur_month != last_month:
            q = (f"A new month of replay just began (UTC month={cur_month}, "
                 f"mid=${mid:,.2f}). Briefly describe the field's current "
                 f"structural state as the new month opens.")
            fw.enqueue("month_open", int(row["minute"]), {
                "month": int(cur_month), "mid_price": mid,
                "regime_product": rp,
            }, q)
            queried = True
            reason  = "month_open"
        last_month = cur_month

        rows_out.append({
            "minute": int(row["minute"]), "mid_price": mid,
            "regime_product": rp, "asymmetry": asy, "coherence": coh,
            "last_trade_age_s": age,
            "vel_max": vel_max_min,
            "trade_count": float(row.get("trade_count", 0.0)),
            "tc_z": float(row.get("tc_z", 0.0)) if ("tc_z" in row.index) else 0.0,
            "events_fired": json.dumps(evs_fired),
            "fractonaut_queried": queried,
            "query_reason": reason,
        })
        if vel_max_min > 0.30:
            log(f"  [STAB] vel_max={vel_max_min:.3f} > 0.30 at min={int(row['minute'])} "
                f"(tc_z={float(row.get('tc_z', 0.0)):.2f})")

        minute_count += 1
        if minute_count % 500 == 0:
            elapsed = time.time() - t_start
            rate = minute_count / max(elapsed, 1e-6)
            remain = (len(df) - minute_count) / max(rate, 1e-6)
            log(f"  [{minute_count:7d}/{len(df):,}] "
                f"rate={rate:6.1f} min/s  eta={remain/3600:.2f}h  "
                f"rp={rp}  fract: ok={fw.n_ok} q={fw.q.qsize()} drop={fw.n_drop}")

        if minute_count % ROLL_FLUSH_N == 0:
            try:
                pd.DataFrame(rows_out).to_parquet(events_parquet, index=False)
                log(f"  flushed {events_parquet} ({len(rows_out)} rows)")
            except Exception as e:
                log(f"  parquet flush error: {e}")

    # final flush
    if rows_out:
        try:
            pd.DataFrame(rows_out).to_parquet(events_parquet, index=False)
            log(f"[END] wrote {events_parquet} ({len(rows_out)} rows)")
        except Exception as e:
            log(f"[END] parquet write error: {e}")

    # synthesis query
    fw.enqueue("replay_end", int(df.iloc[min(len(df)-1, minute_count-1)]["minute"]), {
        "minutes_replayed": minute_count,
        "wall_seconds":     round(time.time() - t_start, 1),
        "recentres":        total_recentres,
        "fractonaut_ok":    fw.n_ok,
    },
    "Three months of BTC replay just finished. Summarise what you saw "
    "across this run: typical regime_product bands, what triggered "
    "the most distinctive events, and any structural pattern that "
    "stood out across the chronicle.")

    log(f"[END] minutes={minute_count}  wall={time.time()-t_start:.1f}s  "
        f"recentres={total_recentres}  fract_ok={fw.n_ok} err={fw.n_err} drop={fw.n_drop}"
        + (f"  virtual_skipped={n_virtual_skipped}" if args.virtual_recentre > 0 else ""))
    log("[END] draining Fractonaut queue (up to 10 min)...")
    fw.shutdown(drain_timeout_s=600)

    if not args.skip_correlation and rows_out:
        try:
            compute_correlations(events_parquet, chronicle_path, correl_md, df, log)
        except Exception as e:
            log(f"[CORR] error: {e}")

    try:
        pub.close(); sub.close(); ctx.term()
    except Exception:
        pass
    log("done.")


if __name__ == "__main__":
    main()