tools: pi_wake_agent.py — Reusable multi-agent wake timer with self-cron/daemon/succession

- Modes: --install, --once, --daemon, --succession, --run, --remove, --list, --status, --validate
- Multi-session support (--session / --sessions)
- Non-blocking h5i bus messages (fire-and-forget)
- Self-cleaning succession mode (--succession --count N --interval X)
- 38 tests passing

pi_wake_agent.py

@@ -0,0 +1,454 @@
+#!/usr/bin/env python3
+"""
+pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession
+
+Usage:
+    pi_wake_agent.py --install --interval 1h --session cc_UV_dev0_Fb --msg "Operator says CONTINUE. Pi here!"
+    pi_wake_agent.py --once --interval 2h --session cc_UV_dev0_Fb --msg "Time's up!"
+    pi_wake_agent.py --daemon --interval 1h --session cc_UV_dev0_Fb
+    pi_wake_agent.py --succession --count 3 --interval 1h --session cc_UV_dev0_Fb --msg "Scheduled wake"
+    pi_wake_agent.py --remove --session cc_UV_dev0_Fb --interval 1h
+    pi_wake_agent.py --list
+    pi_wake_agent.py --status
+    pi_wake_agent.py --validate --session cc_UV_dev0_Fb
+"""
+
+import argparse
+import logging
+import os
+import re
+import shlex
+import signal
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+from typing import List, Optional
+
+# ─── Constants ────────────────────────────────────────────────────────────
+SCRIPT_PATH = Path(__file__).resolve()
+LOG_FILE = Path("/tmp/pi_wake_agent.log")
+LOG_MAX_SIZE = 10 * 1024 * 1024  # 10 MB
+LOG_MAX_FILES = 5
+CRON_COMMENT_PREFIX = "pi_wake_agent"
+AGENT_NICK = "pi_nvnemo"
+H5I_AGENT = "pi_nvnemo"
+H5I_BUS_ROOT = Path("/mnt/dolphinng5_predict")
+DEFAULT_INTERVAL = "1h"
+
+# ─── Logging Setup ───────────────────────────────────────────────────────
+def setup_logging(debug: bool = False) -> logging.Logger:
+    log_rotate()
+    logger = logging.getLogger("pi_wake_agent")
+    logger.setLevel(logging.DEBUG if debug else logging.INFO)
+    
+    fh = logging.FileHandler(LOG_FILE)
+    fh.setLevel(logging.DEBUG)
+    fh.setFormatter(logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
+    logger.addHandler(fh)
+    
+    ch = logging.StreamHandler(sys.stderr)
+    ch.setLevel(logging.WARNING)
+    ch.setFormatter(logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
+    logger.addHandler(ch)
+    
+    return logger
+
+
+def log_rotate() -> None:
+    if not LOG_FILE.exists():
+        return
+    size = LOG_FILE.stat().st_size
+    if size < LOG_MAX_SIZE:
+        return
+    for i in range(LOG_MAX_FILES - 1, 0, -1):
+        src = LOG_FILE.with_suffix(f".log.{i}") if i > 1 else LOG_FILE.with_suffix(".log.1")
+        if src.exists():
+            dst = LOG_FILE.with_suffix(f".log.{i + 1}")
+            src.rename(dst)
+    LOG_FILE.rename(LOG_FILE.with_suffix(".log.1"))
+
+
+# ─── Interval Parsing ────────────────────────────────────────────────────
+def parse_interval(interval: str) -> int:
+    """Parse interval string (e.g., '1h', '30m', '90m', '2h', '10s') to seconds."""
+    match = re.match(r"^(\d+)([hms])$", interval)
+    if not match:
+        raise ValueError(f"Invalid interval format '{interval}'. Use like 1h, 30m, 90m, 2h, 10s")
+    value, unit = int(match.group(1)), match.group(2)
+    if unit == "h":
+        return value * 3600
+    elif unit == "m":
+        return value * 60
+    elif unit == "s":
+        return value
+    raise ValueError(f"Unknown unit: {unit}")
+
+
+def interval_to_cron(interval: str) -> str:
+    """Convert interval to cron schedule."""
+    match = re.match(r"^(\d+)([hm])$", interval)
+    if not match:
+        raise ValueError(f"Cron only supports minutes/hours intervals: {interval}")
+    value, unit = int(match.group(1)), match.group(2)
+    if unit == "h":
+        return f"0 */{value} * * *"
+    elif unit == "m":
+        if value >= 60:
+            raise ValueError(f"For minutes >= 60, use hours (e.g., 1h not 60m)")
+        return f"*/{value} * * * *"
+    raise ValueError(f"Cron only supports minutes/hours intervals: {interval}")
+
+
+def interval_to_human(interval: str) -> str:
+    match = re.match(r"^(\d+)([hms])$", interval)
+    if not match:
+        return interval
+    value, unit = match.group(1), match.group(2)
+    if unit == "h":
+        return f"{value} hour(s)"
+    elif unit == "m":
+        return f"{value} minute(s)"
+    elif unit == "s":
+        return f"{value} second(s)"
+    return interval
+
+
+# ─── Cron Management ────────────────────────────────────────────────────
+def cron_comment(sessions: List[str], interval: str) -> str:
+    sessions_str = ",".join(sessions)
+    return f"{CRON_COMMENT_PREFIX}:{sessions_str}:{interval}"
+
+
+def install_cron(sessions: List[str], interval: str, message: str, logger: logging.Logger) -> None:
+    cron_sched = interval_to_cron(interval)
+    comment = cron_comment(sessions, interval)
+    sessions_arg = f"'{','.join(sessions)}'"
+    cmd = f"cd {H5I_BUS_ROOT} && export H5I_AGENT={H5I_AGENT} && {SCRIPT_PATH} --run --sessions {sessions_arg} --msg {shlex.quote(message)}"
+    
+    result = subprocess.run(["crontab", "-l"], capture_output=True, text=True)
+    existing = result.stdout if result.returncode == 0 else ""
+    lines = [line for line in existing.splitlines() if comment not in line]
+    lines.append(f"{cron_sched} {cmd} # {comment}")
+    new_cron = "\n".join(lines) + "\n"
+    subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True)
+    logger.info(f"Installed cron: {cron_sched} -> {sessions} every {interval}")
+
+
+def remove_cron(sessions: List[str], interval: str, logger: logging.Logger) -> None:
+    comment = cron_comment(sessions, interval)
+    result = subprocess.run(["crontab", "-l"], capture_output=True, text=True)
+    existing = result.stdout if result.returncode == 0 else ""
+    lines = [line for line in existing.splitlines() if comment not in line]
+    new_cron = "\n".join(lines) + ("\n" if lines else "")
+    subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True)
+    logger.info(f"Removed cron for {sessions} ({interval})")
+
+
+def list_cron(logger: logging.Logger) -> None:
+    print("=== pi_wake_agent cron entries ===")
+    result = subprocess.run(["crontab", "-l"], capture_output=True, text=True)
+    existing = result.stdout if result.returncode == 0 else ""
+    found = False
+    for line in existing.splitlines():
+        if CRON_COMMENT_PREFIX in line:
+            print(f"  {line}")
+            found = True
+    if not found:
+        print("  (none)")
+
+
+# ─── Zellij Operations ──────────────────────────────────────────────────
+def zellij_session_exists(session: str) -> bool:
+    try:
+        result = subprocess.run(["zellij", "list-sessions"], capture_output=True, text=True, timeout=5)
+        clean = re.sub(r'\x1b\[[0-9;]*m', '', result.stdout)
+        for line in clean.splitlines():
+            if line.startswith(session + " ") or line == session:
+                return True
+    except Exception:
+        pass
+    return False
+
+
+def zellij_write_chars(session: str, text: str) -> bool:
+    try:
+        subprocess.run(["zellij", "--session", session, "action", "write-chars", text], 
+                       capture_output=True, timeout=5)
+        return True
+    except Exception:
+        return False
+
+
+def zellij_write_enter(session: str) -> bool:
+    try:
+        subprocess.run(["zellij", "--session", session, "action", "write", "13"], 
+                       capture_output=True, timeout=5)
+        return True
+    except Exception:
+        return False
+
+
+# ─── Wake Action ────────────────────────────────────────────────────────────────────
+    logger.info(f"One-shot timer set for {interval} ({interval_to_human(interval)})")
+    
+    def _wake():
+        time.sleep(interval_seconds)
+        run_wake(sessions, message, logger)
+    
+    thread = threading.Thread(target=_wake, daemon=True)
+    thread.start()
+    logger.info(f"Background timer started (thread: {thread.ident})")
+    pid_file = Path(f"/tmp/pi_wake_agent_{sessions[0]}.pid")
+    pid_file.write_text(str(os.getpid()))
+
+
+def run_wake(sessions: List[str], message: str, logger: logging.Logger) -> None:
+    sessions_str = ",".join(sessions)
+    logger.info(f"Waking {sessions_str} with message: {message}")
+    
+    for session in sessions:
+        zellij_write_chars(session, f"[{AGENT_NICK} via zellij] {message} Run: h5i-bus msg inbox")
+        for _ in range(5):
+            zellij_write_enter(session)
+            time.sleep(1)
+    
+    # Fire-and-forget h5i bus message (non-blocking)
+    def _send_bus():
+        try:
+            subprocess.run(
+                ["h5i", "msg", "send", "Fable", f"{message} (timer wakeup)"],
+                cwd=H5I_BUS_ROOT,
+                env={**os.environ, "H5I_AGENT": H5I_AGENT},
+                capture_output=True,
+                timeout=5
+            )
+        except Exception:
+            pass  # Silently ignore - fire and forget
+    
+    threading.Thread(target=_send_bus, daemon=True).start()
+    
+    logger.info(f"Wake sent to {sessions_str}")
+
+def run_once(sessions: List[str], interval: str, message: str, logger: logging.Logger) -> None:
+    interval_seconds = parse_interval(interval)
+    logger.info(f"One-shot timer set for {interval} ({interval_to_human(interval)})")
+    
+    def _wake():
+        time.sleep(interval_seconds)
+        run_wake(sessions, message, logger)
+    
+    thread = threading.Thread(target=_wake, daemon=True)
+    thread.start()
+    logger.info(f"Background timer started (thread: {thread.ident})")
+    pid_file = Path(f"/tmp/pi_wake_agent_{sessions[0]}.pid")
+    pid_file.write_text(str(os.getpid()))
+    return 0
+
+
+def run_daemon(sessions: List[str], interval: str, message: str, logger: logging.Logger) -> None:
+    interval_seconds = parse_interval(interval)
+    logger.info("=== DAEMON START ===")
+    logger.info(f"Interval: {interval} ({interval_to_human(interval)})")
+    logger.info(f"Sessions: {sessions}")
+    logger.info(f"Message: {message}")
+    
+    run_wake(sessions, message, logger)
+    
+    while True:
+        logger.debug(f"Sleeping for {interval_seconds}s...")
+        time.sleep(interval_seconds)
+        run_wake(sessions, message, logger)
+
+
+def run_succession(sessions: List[str], interval: str, count: int, message: str, logger: logging.Logger) -> None:
+    """Run wake N times at interval, then self-clean (remove cron if installed)."""
+    interval_seconds = parse_interval(interval)
+    logger.info(f"=== SUCCESSION START === Count: {count}, Interval: {interval} ({interval_to_human(interval)})")
+    logger.info(f"Sessions: {sessions}")
+    logger.info(f"Message: {message}")
+    
+    for i in range(1, count + 1):
+        logger.info(f"Succession {i}/{count}")
+        run_wake(sessions, message, logger)
+        if i < count:
+            logger.debug(f"Sleeping for {interval_seconds}s until next succession...")
+            time.sleep(interval_seconds)
+    
+    # Self-clean: remove any cron entry for this session/interval combo
+    try:
+        remove_cron(sessions, interval, logger)
+        logger.info("Self-cleanup complete (cron removed)")
+    except Exception as e:
+        logger.warning(f"Self-cleanup failed: {e}")
+    
+    logger.info("=== SUCCESSION COMPLETE ===")
+
+
+def status(logger: logging.Logger) -> None:
+    print("=== pi_wake_agent Status ===")
+    print(f"Script: {SCRIPT_PATH}")
+    print(f"Log: {LOG_FILE}")
+    print(f"Log size: {LOG_FILE.stat().st_size if LOG_FILE.exists() else 'N/A'} bytes")
+    print(f"Agent: {AGENT_NICK}")
+    print()
+    list_cron(logger)
+    print()
+    print("=== Active one-shot timers ===")
+    found = False
+    for pid_file in Path("/tmp").glob("pi_wake_agent_*.pid"):
+        found = True
+        try:
+            pid = int(pid_file.read_text().strip())
+            os.kill(pid, 0)
+            print(f"  PID {pid} (active)")
+        except (ProcessLookupError, ValueError):
+            print(f"  PID {pid_file.read_text().strip()} (dead, cleaning up)")
+            pid_file.unlink(missing_ok=True)
+    if not found:
+        print("  (none)")
+
+
+def validate_sessions(sessions: List[str], logger: logging.Logger) -> None:
+    logger.info(f"Validating sessions: {sessions}")
+    for session in sessions:
+        if zellij_session_exists(session):
+            logger.info(f"  {session}: EXISTS")
+        else:
+            logger.warning(f"  {session}: NOT FOUND (may be dead/EXITED)")
+
+
+# ─── Session Parsing ─────────────────────────────────────────────────────
+def parse_sessions(raw) -> List[str]:
+    if not raw:
+        return []
+    if isinstance(raw, list):
+        return [s.strip() for s in raw]
+    return [s.strip() for s in raw.split(",") if s.strip()]
+
+
+# ─── Usage ──────────────────────────────────────────────────────────────
+def create_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+EXAMPLES:
+  # Install recurring 1-hour timer for one session
+  pi_wake_agent.py --install --interval 1h --session cc_UV_dev0_Fb
+
+  # Install 30-minute timer for multiple sessions
+  pi_wake_agent.py --install --interval 30m --sessions "cc_UV_dev0_Fb,cc_UV_dev1_48" --msg "Wake up!"
+
+  # One-shot wake in 2 hours (no cron)
+  pi_wake_agent.py --once --interval 2h --session cc_UV_dev0_Fb --msg "Time's up!"
+
+  # Run N times at interval, then self-clean
+  pi_wake_agent.py --succession --count 3 --interval 1h --session cc_UV_dev0_Fb --msg "Scheduled wake"
+
+  # Run as daemon (long-lived process, no cron)
+  pi_wake_agent.py --daemon --interval 1h --session cc_UV_dev0_Fb
+
+  # Remove timer
+  pi_wake_agent.py --remove --session cc_UV_dev0_Fb --interval 1h
+
+  # List all timers
+  pi_wake_agent.py --list
+
+  # Show status
+  pi_wake_agent.py --status
+
+  # Validate sessions
+  pi_wake_agent.py --validate --session cc_UV_dev0_Fb
+"""
+    )
+    parser = argparse.ArgumentParser(
+        description="pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--install", action="store_const", const="install", dest="mode", help="Install recurring cron timer")
+    parser.add_argument("--once", action="store_const", const="once", dest="mode", help="One-shot wake (no cron)")
+    parser.add_argument("--daemon", action="store_const", const="daemon", dest="mode", help="Run as long-lived daemon (no cron)")
+    parser.add_argument("--succession", action="store_const", const="succession", dest="mode", help="Run N times at interval, then self-clean")
+    parser.add_argument("--count", type=int, default=1, help="Number of successions (for --succession mode)")
+    parser.add_argument("--run", action="store_const", const="run", dest="mode", help="Internal: run wake action (called by cron)")
+    parser.add_argument("--remove", action="store_const", const="remove", dest="mode", help="Remove cron timer")
+    parser.add_argument("--list", action="store_const", const="list", dest="mode", help="List active cron timers")
+    parser.add_argument("--status", action="store_const", const="status", dest="mode", help="Show status (cron + one-shot timers)")
+    parser.add_argument("--validate", action="store_const", const="validate", dest="mode", help="Validate sessions exist in zellij")
+    parser.add_argument("--interval", default=DEFAULT_INTERVAL, help="Interval (default: 1h). Formats: 30m, 1h, 90m, 2h, etc.")
+    parser.add_argument("--session", action="append", dest="session_list", help="Zellij session name (can repeat)")
+    parser.add_argument("--sessions", help="Comma-separated list of sessions")
+    parser.add_argument("--msg", default=f"Operator says CONTINUE. {AGENT_NICK} here, saying hi!", help="Wake message")
+    parser.add_argument("--debug", action="store_true", help="Enable debug logging")
+    parser.set_defaults(mode="install")
+    return parser
+
+
+# ─── Main ───────────────────────────────────────────────────────────────
+def main() -> int:
+    parser = create_parser()
+    args = parser.parse_args()
+    
+    # Combine sessions
+    sessions: List[str] = []
+    if args.sessions:
+        sessions.extend(parse_sessions(args.sessions))
+    if args.session_list:
+        sessions.extend(args.session_list)
+    
+    # Setup logging
+    logger = setup_logging(args.debug)
+    
+    # Validate
+    if len(sessions) == 0 and args.mode not in ("list", "status"):
+        logger.error("--session or --sessions required")
+        parser.print_help()
+        return 1
+    
+    if args.mode == "succession" and args.count < 1:
+        logger.error("--count must be >= 1 for succession mode")
+        return 1
+    
+    logger.info(f"=== pi_wake_agent {args.mode} ===")
+    logger.info(f"Interval: {args.interval} ({interval_to_human(args.interval)})")
+    logger.info(f"Sessions: {sessions}")
+    logger.info(f"Message: {args.msg}")
+    if args.mode == "succession":
+        logger.info(f"Count: {args.count}")
+    
+    try:
+        if args.mode == "install":
+            install_cron(sessions, args.interval, args.msg, logger)
+        elif args.mode == "once":
+            run_once(sessions, args.interval, args.msg, logger)
+        elif args.mode == "daemon":
+            run_daemon(sessions, args.interval, args.msg, logger)
+        elif args.mode == "succession":
+            run_succession(sessions, args.interval, args.count, args.msg, logger)
+        elif args.mode == "run":
+            run_wake(sessions, args.msg, logger)
+        elif args.mode == "remove":
+            remove_cron(sessions, args.interval, logger)
+        elif args.mode == "list":
+            list_cron(logger)
+        elif args.mode == "status":
+            status(logger)
+        elif args.mode == "validate":
+            validate_sessions(sessions, logger)
+        else:
+            logger.error(f"Unknown mode: {args.mode}")
+            return 1
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        if args.debug:
+            import traceback
+            traceback.print_exc()
+        return 1
+    
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

prod/ch_writer.py

@@ -60,6 +60,13 @@ CH_WAL_TRUNCATE_BYTES = int(os.environ.get("CH_WAL_TRUNCATE_BYTES", str(64 * 102
 CH_VACUUM_MIN_BYTES = int(os.environ.get("CH_VACUUM_MIN_BYTES", str(512 * 1024 * 1024)))
 CH_VACUUM_MIN_FREE_RATIO = float(os.environ.get("CH_VACUUM_MIN_FREE_RATIO", "1.25"))
 CH_VACUUM_MIN_FREE_BYTES = int(os.environ.get("CH_VACUUM_MIN_FREE_BYTES", str(128 * 1024 * 1024)))
+# Poison-row quarantine: a row CH permanently rejects (schema mismatch, bad
+# value) must not head-of-line-block the spool forever. After this many
+# failed attempts the row is retried INDIVIDUALLY (CH proven up first); if it
+# still fails it moves to the dead_letter table for offline repair/replay.
+# Incident 2026-06-12: one trade_events row with bars_held=-106 (UInt16
+# column) was retried 3.2M times and jammed 18M rows behind it for 1.5 days.
+CH_POISON_ATTEMPTS = int(os.environ.get("CH_POISON_ATTEMPTS", "200"))
 
 
 # ─── Timestamp helpers ────────────────────────────────────────────────────────
@@ -202,6 +209,19 @@ class _CHWriter:
         conn.execute(
             "CREATE INDEX IF NOT EXISTS idx_queue_table ON queue(table_name, id)"
         )
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS dead_letter (
+                id INTEGER PRIMARY KEY,
+                table_name TEXT NOT NULL,
+                payload TEXT NOT NULL,
+                created_ts_us INTEGER NOT NULL,
+                attempts INTEGER NOT NULL,
+                dead_ts_us INTEGER NOT NULL,
+                last_error TEXT
+            )
+            """
+        )
         return conn
 
     _PUT_LOCK_TIMEOUT_S: float = 0.1  # max wait before dropping the row
@@ -276,7 +296,7 @@ class _CHWriter:
         with self._lock:
             cur = self._conn.execute(
                 "SELECT id, attempts FROM queue WHERE id IN (%s)" % ",".join("?" for _ in ids),
-                [int(_id) for _ in ids],
+                [int(_id) for _id in ids],
             )
             high_attempts = [(row[0], int(row[1]) + 1) for row in cur.fetchall() if int(row[1]) >= 1000]
             self._conn.executemany(
@@ -291,6 +311,65 @@ class _CHWriter:
                     row_id, attempt,
                 )
 
+    def _ch_alive(self) -> bool:
+        """True iff ClickHouse answers a trivial query — used to distinguish
+        'CH is down' (retry forever, quarantine nothing) from 'CH rejects this
+        specific row' (quarantine after CH_POISON_ATTEMPTS)."""
+        try:
+            req = urllib.request.Request(f"{CH_URL}/?query=SELECT+1", method="GET")
+            req.add_header("X-ClickHouse-User", CH_USER)
+            req.add_header("X-ClickHouse-Key", CH_PASS)
+            with urllib.request.urlopen(req, timeout=3) as resp:
+                return resp.status == 200
+        except Exception:
+            return False
+
+    def _quarantine_poison(self, table: str, items: List[Tuple[int, dict]]) -> None:
+        """After a batch failure, isolate rows CH permanently rejects.
+
+        Only rows whose attempt count exceeds CH_POISON_ATTEMPTS are touched,
+        and only while CH itself is provably up. Each candidate is retried
+        alone: success → delivered+deleted; failure → moved to dead_letter
+        (payload preserved for offline repair/replay) so the spool can drain.
+        """
+        ids = [row_id for row_id, _ in items]
+        if not ids:
+            return
+        with self._lock:
+            cur = self._conn.execute(
+                "SELECT id, attempts FROM queue WHERE id IN (%s)"
+                % ",".join("?" for _ in ids),
+                [int(i) for i in ids],
+            )
+            attempts_by_id = {int(r[0]): int(r[1]) for r in cur.fetchall()}
+        candidates = [
+            (row_id, payload) for row_id, payload in items
+            if attempts_by_id.get(int(row_id), 0) >= CH_POISON_ATTEMPTS
+        ]
+        if not candidates:
+            return
+        if not self._ch_alive():
+            return  # CH outage — nothing is poison, keep retrying the batch
+        for row_id, payload in candidates:
+            if self._post_rows(table, [payload]):
+                self._delete_ids([row_id])
+                continue
+            now = ts_us()
+            with self._lock:
+                self._conn.execute(
+                    "INSERT OR REPLACE INTO dead_letter "
+                    "(id, table_name, payload, created_ts_us, attempts, dead_ts_us, last_error) "
+                    "SELECT id, table_name, payload, created_ts_us, attempts, ?, ? "
+                    "FROM queue WHERE id=?",
+                    (now, "rejected by CH while CH alive (see ch flush WARNINGs)", int(row_id)),
+                )
+                self._conn.execute("DELETE FROM queue WHERE id=?", (int(row_id),))
+            log.error(
+                "ch_writer[%s]: POISON ROW quarantined to dead_letter: id=%s table=%s "
+                "attempts=%d — spool unblocked; repair/replay offline",
+                self._db, row_id, table, attempts_by_id.get(int(row_id), -1),
+            )
+
     def _queue_count(self) -> int:
         with self._lock:
             row = self._conn.execute("SELECT count(*) FROM queue").fetchone()
@@ -433,7 +512,7 @@ class _CHWriter:
             raw = resp.read().decode("utf-8", errors="replace")
             return [line for line in raw.splitlines() if line]
 
-    def _existing_trade_keys(self, rows: List[dict]) -> set[Tuple[str, int]]:
+    def _existing_trade_keys(self, rows: List[dict]) -> set[Tuple[str, str]]:
         trade_ids: List[str] = []
         for row in rows:
             trade_id = row.get("trade_id")
@@ -447,13 +526,14 @@ class _CHWriter:
             return set()
 
         unique = sorted(set(trade_ids))
-        existing: set[Tuple[str, int]] = set()
+        existing: set[Tuple[str, str]] = set()
         chunk_size = 200
         for i in range(0, len(unique), chunk_size):
             chunk = unique[i : i + chunk_size]
             quoted = ",".join("'" + tid.replace("'", "''") + "'" for tid in chunk)
             sql = (
-                "SELECT trade_id, toInt64(toUnixTimestamp64Micro(ts)) "
+                "SELECT trade_id, "
+                "ifNull(nullIf(event_id, ''), concat(toString(toInt64(toUnixTimestamp64Micro(ts))), ':', ifNull(exit_reason, ''))) "
                 f"FROM trade_events WHERE trade_id IN ({quoted}) FORMAT TSV"
             )
             try:
@@ -466,13 +546,24 @@ class _CHWriter:
                 parts = line.split("\t", 1)
                 if len(parts) != 2:
                     continue
-                tid, ts_us_s = parts
+                tid, event_key = parts
-                try:
+                existing.add((tid, event_key))
-                    existing.add((tid, int(ts_us_s)))
-                except Exception:
-                    continue
         return existing
 
+    @staticmethod
+    def _trade_event_key(payload: dict) -> Tuple[str, str] | None:
+        tid = str(payload.get("trade_id", "") or "").strip()
+        if not tid:
+            return None
+        event_id = str(payload.get("event_id", "") or "").strip()
+        if event_id:
+            return (tid, event_id)
+        try:
+            ts_us_val = int(payload.get("ts"))
+        except Exception:
+            ts_us_val = -1
+        return (tid, f"{ts_us_val}:{payload.get('exit_reason', '')}")
+
     def flush_once(self) -> int:
         """
         Drain a single batch from the local spool.
@@ -494,38 +585,37 @@ class _CHWriter:
             rows = [payload for _, payload in items]
             if table == "trade_events":
                 existing = self._existing_trade_keys(rows)
-                if existing:
+                seen = set(existing)
-                    kept_ids: List[int] = []
+                kept_ids: List[int] = []
-                    kept_rows: List[dict] = []
+                kept_rows: List[dict] = []
-                    duplicate_ids: List[int] = []
+                duplicate_ids: List[int] = []
-                    for row_id, payload in items:
+                for row_id, payload in items:
-                        tid = str(payload.get("trade_id", "")).strip()
+                    probe = self._trade_event_key(payload)
-                        try:
+                    if probe is not None and probe in seen:
-                            ts_us_val = int(payload.get("ts"))
+                        duplicate_ids.append(row_id)
-                        except Exception:
-                            ts_us_val = -1
-                        if tid and ts_us_val >= 0 and (tid, ts_us_val) in existing:
-                            duplicate_ids.append(row_id)
-                        else:
-                            kept_ids.append(row_id)
-                            kept_rows.append(payload)
-                    if duplicate_ids:
-                        self._delete_ids(duplicate_ids)
-                        log.warning(
-                            "ch_writer[%s]: dropped %d duplicate trade_events rows by trade_id",
-                            self._db,
-                            len(duplicate_ids),
-                        )
-                    ids = kept_ids
-                    rows = kept_rows
-                    if not rows:
                         continue
+                    kept_ids.append(row_id)
+                    kept_rows.append(payload)
+                    if probe is not None:
+                        seen.add(probe)
+                if duplicate_ids:
+                    self._delete_ids(duplicate_ids)
+                    log.warning(
+                        "ch_writer[%s]: dropped %d duplicate trade_events rows by stable event key",
+                        self._db,
+                        len(duplicate_ids),
+                    )
+                ids = kept_ids
+                rows = kept_rows
+                if not rows:
+                    continue
             ok = self._post_rows(table, rows)
             if ok:
                 delivered += len(rows)
                 self._delete_ids(ids)
             else:
                 self._bump_attempts(ids)
+                self._quarantine_poison(table, list(zip(ids, rows)))
         self._maybe_maintain_spool()
         return delivered
 

prod/clean_arch/dita_v2/__init__.py

@@ -36,8 +36,6 @@ from .contracts import (
 )
 from .journal import ClickHouseKernelJournal, KernelJournal, MemoryKernelJournal
 from .rust_backend import ExecutionKernel
-from .bingx_venue import BingxVenueAdapter
-from .launcher import DITAv2LauncherBundle, LauncherVenueMode, LauncherZincMode, build_launcher_bundle
 from .projection import HazelcastProjection, build_position_state_row, build_projection
 from .venue import VenueAdapter
 from .mock_venue import MockVenueAdapter, MockVenueScenario
@@ -45,6 +43,28 @@ from .zinc_plane import InMemoryZincPlane, ZincPlane
 from .real_zinc_plane import RealZincPlane, RealZincUnavailable
 from .real_control_plane import RealZincControlPlane, RealZincUnavailable as RealZincControlUnavailable
 
+
+def __getattr__(name: str):
+    if name == "BingxVenueAdapter":
+        from .bingx_venue import BingxVenueAdapter
+
+        return BingxVenueAdapter
+    if name in {"DITAv2LauncherBundle", "LauncherVenueMode", "LauncherZincMode", "build_launcher_bundle"}:
+        from .launcher import (
+            DITAv2LauncherBundle,
+            LauncherVenueMode,
+            LauncherZincMode,
+            build_launcher_bundle,
+        )
+
+        return {
+            "DITAv2LauncherBundle": DITAv2LauncherBundle,
+            "LauncherVenueMode": LauncherVenueMode,
+            "LauncherZincMode": LauncherZincMode,
+            "build_launcher_bundle": build_launcher_bundle,
+        }[name]
+    raise AttributeError(name)
+
 __all__ = [
     "AccountProjection",
     "AccountSnapshot",

prod/clean_arch/dita_v2/account.py

@@ -8,6 +8,7 @@ from enum import Enum
 from typing import Any, Dict, Iterable, List, Optional
 import math
 import time
+import threading
 
 from .contracts import TradeSide, TradeSlot, TradeStage
 from .utils import safe_float
@@ -59,42 +60,49 @@ class AccountProjection:
         GIL guarantees single-field reference assignment is atomic, so readers
         that hold snap = kernel.account.snapshot before use see a consistent view.
         """
-        cur = self.snapshot
+        with self._lock:
-        self.snapshot = AccountSnapshot(
+            cur = self.snapshot
-            capital=kw.get("capital", cur.capital),
+            self.snapshot = AccountSnapshot(
-            equity=kw.get("equity", cur.equity),
+                capital=kw.get("capital", cur.capital),
-            realized_pnl=kw.get("realized_pnl", cur.realized_pnl),
+                equity=kw.get("equity", cur.equity),
-            unrealized_pnl=kw.get("unrealized_pnl", cur.unrealized_pnl),
+                realized_pnl=kw.get("realized_pnl", cur.realized_pnl),
-            open_positions=kw.get("open_positions", cur.open_positions),
+                unrealized_pnl=kw.get("unrealized_pnl", cur.unrealized_pnl),
-            open_notional=kw.get("open_notional", cur.open_notional),
+                open_positions=kw.get("open_positions", cur.open_positions),
-            fees_paid=kw.get("fees_paid", cur.fees_paid),
+                open_notional=kw.get("open_notional", cur.open_notional),
-            trade_seq=kw.get("trade_seq", cur.trade_seq),
+                fees_paid=kw.get("fees_paid", cur.fees_paid),
-            peak_capital=kw.get("peak_capital", cur.peak_capital),
+                trade_seq=kw.get("trade_seq", cur.trade_seq),
-            capital_source=kw.get("capital_source", cur.capital_source),
+                peak_capital=kw.get("peak_capital", cur.peak_capital),
-            e_wallet_balance=kw.get("e_wallet_balance", cur.e_wallet_balance),
+                capital_source=kw.get("capital_source", cur.capital_source),
-            event_seq=kw.get("event_seq", cur.event_seq),
+                e_wallet_balance=kw.get("e_wallet_balance", cur.e_wallet_balance),
-        )
+                event_seq=kw.get("event_seq", cur.event_seq),
+            )
+
+    def __post_init__(self) -> None:
+        self._lock = threading.RLock()
 
     def observe_slots(self, slots: Iterable[TradeSlot]) -> None:
-        open_positions = 0
+        with self._lock:
-        open_notional = 0.0
+            open_positions = 0
-        unrealized_pnl = 0.0
+            open_notional = 0.0
-        for slot in slots:
+            unrealized_pnl = 0.0
-            if slot.closed or slot.size <= 0:
+            for slot in slots:
-                continue
+                if slot.closed or slot.size <= 0:
-            if slot.fsm_state in {TradeStage.POSITION_OPEN, TradeStage.POSITION_OPENED, TradeStage.ENTRY_WORKING, TradeStage.EXIT_WORKING}:
+                    continue
-                open_positions += 1
+                if slot.fsm_state in {TradeStage.POSITION_OPEN, TradeStage.POSITION_OPENED, TradeStage.ENTRY_WORKING, TradeStage.EXIT_WORKING}:
-                mark = safe_float(slot.entry_price, 0.0)
+                    open_positions += 1
-                mark = safe_float(slot.metadata.get("mark_price"), mark)
+                    mark = safe_float(slot.entry_price, 0.0)
-                open_notional += abs(slot.size) * abs(mark)
+                    mark = safe_float(slot.metadata.get("mark_price"), mark)
-                unrealized_pnl += float(slot.unrealized_pnl or 0.0)
+                    open_notional += abs(slot.size) * abs(mark)
-        self._replace_snapshot(
+                    unrealized_pnl += float(slot.unrealized_pnl or 0.0)
-            open_positions=open_positions,
+            capital = self.snapshot.capital
-            open_notional=open_notional,
+            peak_capital = self.snapshot.peak_capital
-            unrealized_pnl=unrealized_pnl,
+            self._replace_snapshot(
-            equity=self.snapshot.capital + unrealized_pnl if math.isfinite(self.snapshot.capital + unrealized_pnl) else self.snapshot.capital,
+                open_positions=open_positions,
-            peak_capital=max(self.snapshot.peak_capital, self.snapshot.capital) if open_notional > 0 and self.snapshot.capital > 0 else self.snapshot.peak_capital,
+                open_notional=open_notional,
-        )
+                unrealized_pnl=unrealized_pnl,
+                equity=capital + unrealized_pnl if math.isfinite(capital + unrealized_pnl) else capital,
+                peak_capital=max(peak_capital, capital) if open_notional > 0 and capital > 0 else peak_capital,
+            )
 
     def anchor_to_exchange(self, wallet_balance: float, available_margin: float, event_seq: int) -> None:
         """Snap published capital to exchange wallet balance.
@@ -106,39 +114,42 @@ class AccountProjection:
         Guards: wallet_balance must be > 0 and finite (the zero-wb frame lesson
         from ACCOUNT_UPDATE frames with no USDT balance entry).
         """
-        wb = safe_float(wallet_balance, 0.0)
+        with self._lock:
-        if wb <= 0.0 or not math.isfinite(wb):
+            wb = safe_float(wallet_balance, 0.0)
-            return
+            if wb <= 0.0 or not math.isfinite(wb):
-        self.snapshot.capital = wb
+                return
-        self.snapshot.e_wallet_balance = wb
+            self.snapshot.capital = wb
-        self.snapshot.capital_source = "e_anchored"
+            self.snapshot.e_wallet_balance = wb
-        self.snapshot.event_seq = int(event_seq)
+            self.snapshot.capital_source = "e_anchored"
-        self.snapshot.equity = wb + self.snapshot.unrealized_pnl
+            self.snapshot.event_seq = int(event_seq)
-        if not math.isfinite(self.snapshot.equity):
+            self.snapshot.equity = wb + self.snapshot.unrealized_pnl
-            self.snapshot.equity = wb
+            if not math.isfinite(self.snapshot.equity):
-        self.snapshot.peak_capital = max(self.snapshot.peak_capital, wb)
+                self.snapshot.equity = wb
+            self.snapshot.peak_capital = max(self.snapshot.peak_capital, wb)
 
     def settle(self, realized_pnl: float, fees: float = 0.0) -> None:
-        rp = safe_float(realized_pnl, 0.0)
+        with self._lock:
-        # Include fees in capital delta (today fees only accumulate in
+            cur = self.snapshot
-        # fees_paid while published capital ignores them between reseeds).
+            rp = safe_float(realized_pnl, 0.0)
-        net = rp - safe_float(fees, 0.0)
+            # Include fees in capital delta (today fees only accumulate in
-        new_capital = safe_float(self.snapshot.capital + net, self.snapshot.capital)
+            # fees_paid while published capital ignores them between reseeds).
-        if self.max_capital is not None:
+            net = rp - safe_float(fees, 0.0)
-            new_capital = min(new_capital, self.max_capital)
+            new_capital = safe_float(cur.capital + net, cur.capital)
-        new_capital = max(self.min_capital, new_capital)
+            if self.max_capital is not None:
-        new_source = self.snapshot.capital_source
+                new_capital = min(new_capital, self.max_capital)
-        if new_source == "e_anchored" and abs(net) > 1e-12:
+            new_capital = max(self.min_capital, new_capital)
-            new_source = "k_bridged"
+            new_source = cur.capital_source
-        new_fees = self.snapshot.fees_paid + safe_float(fees, 0.0)
+            if new_source == "e_anchored" and abs(net) > 1e-12:
-        new_equity = new_capital + self.snapshot.unrealized_pnl
+                new_source = "k_bridged"
-        if not math.isfinite(new_equity):
+            new_fees = cur.fees_paid + safe_float(fees, 0.0)
-            new_equity = new_capital
+            new_equity = new_capital + cur.unrealized_pnl
-        self._replace_snapshot(
+            if not math.isfinite(new_equity):
-            capital=new_capital, capital_source=new_source,
+                new_equity = new_capital
-            realized_pnl=self.snapshot.realized_pnl + rp,
+            self._replace_snapshot(
-            fees_paid=new_fees, equity=new_equity,
+                capital=new_capital, capital_source=new_source,
-        )
+                realized_pnl=cur.realized_pnl + rp,
+                fees_paid=new_fees, equity=new_equity,
+            )
 
     def to_account_event(
         self,
@@ -154,31 +165,32 @@ class AccountProjection:
         bars_held: int = 0,
         metadata: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
-        self.snapshot.equity = self.snapshot.capital + self.snapshot.unrealized_pnl
+        with self._lock:
-        return {
+            self.snapshot.equity = self.snapshot.capital + self.snapshot.unrealized_pnl
-            "timestamp": timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp),
+            return {
-            "runtime_namespace": self.runtime_namespace,
+                "timestamp": timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp),
-            "strategy_namespace": self.strategy_namespace,
+                "runtime_namespace": self.runtime_namespace,
-            "event_namespace": self.event_namespace,
+                "strategy_namespace": self.strategy_namespace,
-            "actor_name": self.actor_name,
+                "event_namespace": self.event_namespace,
-            "exec_venue": self.exec_venue,
+                "actor_name": self.actor_name,
-            "data_venue": self.data_venue,
+                "exec_venue": self.exec_venue,
-            "ledger_authority": self.ledger_authority,
+                "data_venue": self.data_venue,
-            "capital": float(self.snapshot.capital),
+                "ledger_authority": self.ledger_authority,
-            "equity": float(self.snapshot.equity),
+                "capital": float(self.snapshot.capital),
-            "open_positions": int(self.snapshot.open_positions),
+                "equity": float(self.snapshot.equity),
-            "current_open_notional": float(self.snapshot.open_notional),
+                "open_positions": int(self.snapshot.open_positions),
-            "current_account_leverage": float(self.snapshot.leverage),
+                "current_open_notional": float(self.snapshot.open_notional),
-            "trade_id": trade_id,
+                "current_account_leverage": float(self.snapshot.leverage),
-            "asset": asset,
+                "trade_id": trade_id,
-            "side": side.value,
+                "asset": asset,
-            "reason": reason,
+                "side": side.value,
-            "stage": stage.value,
+                "reason": reason,
-            "pnl": float(pnl),
+                "stage": stage.value,
-            "pnl_pct": float(pnl_pct),
+                "pnl": float(pnl),
-            "bars_held": int(bars_held),
+                "pnl_pct": float(pnl_pct),
-            "metadata": dict(metadata or {}),
+                "bars_held": int(bars_held),
-        }
+                "metadata": dict(metadata or {}),
+            }
 
 
 # ---------------------------------------------------------------------------
@@ -311,6 +323,7 @@ class AccountProjectionV2:
         self._min_capital = min_capital
         self._max_capital = max_capital
         self._cfg = reconcile_config or ReconcileConfig()
+        self._lock = threading.RLock()
 
         # Running K-value accumulators
         self._k_realized: float = 0.0
@@ -345,16 +358,18 @@ class AccountProjectionV2:
         fee: float,
         realized_pnl: float,
     ) -> None:
-        self._k_realized += _safe(realized_pnl)
+        with self._lock:
-        self._k_fees += _safe(fee)
+            self._k_realized += _safe(realized_pnl)
-        self._e_last_fill_price = _safe(fill_price)
+            self._k_fees += _safe(fee)
-        self._e_last_fill_qty = _safe(fill_qty)
+            self._e_last_fill_price = _safe(fill_price)
-        self._e_last_fill_fee = _safe(fee)
+            self._e_last_fill_qty = _safe(fill_qty)
-        self._e_last_fill_realized = _safe(realized_pnl)
+            self._e_last_fill_fee = _safe(fee)
+            self._e_last_fill_realized = _safe(realized_pnl)
 
     def apply_funding(self, amount: float) -> None:
-        self._k_funding += _safe(amount)
+        with self._lock:
-        self._e_last_funding = _safe(amount)
+            self._k_funding += _safe(amount)
+            self._e_last_funding = _safe(amount)
 
     def apply_balance_update(
         self,
@@ -364,13 +379,15 @@ class AccountProjectionV2:
         used_margin: float,
         maint_margin: float,
     ) -> None:
-        self._e_wallet_balance = _safe(wallet_balance)
+        with self._lock:
-        self._e_avail_margin = _safe(available_margin)
+            self._e_wallet_balance = _safe(wallet_balance)
-        self._e_used_margin = _safe(used_margin)
+            self._e_avail_margin = _safe(available_margin)
-        self._e_maint_margin = _safe(maint_margin)
+            self._e_used_margin = _safe(used_margin)
+            self._e_maint_margin = _safe(maint_margin)
 
     def apply_position_update(self, positions: List[EPosition]) -> None:
-        self._e_positions = list(positions)
+        with self._lock:
+            self._e_positions = list(positions)
 
     # ------------------------------------------------------------------
     # Snapshot construction (called after each ingestion step)
@@ -382,21 +399,24 @@ class AccountProjectionV2:
         slots: Iterable[TradeSlot],
         ts: Optional[float] = None,
     ) -> AccountSnapshotV2:
-        self._event_seq += 1
+        with self._lock:
-        snap = self._build(self._event_seq, source_event_id, list(slots), ts or time.time())
+            self._event_seq += 1
-        self._snapshot = snap
+            snap = self._build(self._event_seq, source_event_id, list(slots), ts or time.time())
-        return snap
+            self._snapshot = snap
+            return snap
 
     @property
     def snapshot(self) -> AccountSnapshotV2:
-        return self._snapshot
+        with self._lock:
+            return self._snapshot
 
     @property
     def k_capital(self) -> float:
-        raw = self._seed + self._k_realized - self._k_fees - self._k_funding
+        with self._lock:
-        if self._max_capital is not None:
+            raw = self._seed + self._k_realized - self._k_fees - self._k_funding
-            raw = min(raw, self._max_capital)
+            if self._max_capital is not None:
-        return max(self._min_capital, raw)
+                raw = min(raw, self._max_capital)
+            return max(self._min_capital, raw)
 
     # ------------------------------------------------------------------
     # Internal helpers

prod/clean_arch/dita_v2/real_control_plane.py

@@ -8,6 +8,8 @@ import sys
 from pathlib import Path
 from typing import Any, Dict, Optional
 
+import threading
+
 from .control import BackendMode, ControlPlane, ControlUpdate, KernelControlSnapshot, KernelMode, KernelVerbosity
 
 _ZINC_ADAPTER_PATH = Path(__file__).resolve().parents[3] / "zinc" / "adapters" / "python"
@@ -70,6 +72,7 @@ class RealZincControlPlane(ControlPlane):
         require_real_zinc()
         base = prefix.strip("/").replace("/", "_")
         self.region_name = f"{base}_control"
+        self._lock = threading.RLock()
         self._seq = 0
         self._snapshot = KernelControlSnapshot()
         if create:
@@ -86,21 +89,24 @@ class RealZincControlPlane(ControlPlane):
         self.region.close()
 
     def read(self) -> KernelControlSnapshot:
-        payload = _decode_packet(self.region.as_buffer())
+        with self._lock:
-        control = payload.get("control") if isinstance(payload, dict) else None
+            payload = _decode_packet(self.region.as_buffer())
-        if not isinstance(control, dict):
+            control = payload.get("control") if isinstance(payload, dict) else None
+            if not isinstance(control, dict):
+                return self._snapshot
+            self._snapshot = KernelControlSnapshot(**control)
             return self._snapshot
-        self._snapshot = KernelControlSnapshot(**control)
-        return self._snapshot
 
     def update(self, update: ControlUpdate) -> KernelControlSnapshot:
-        self._snapshot = update.apply(self.read())
+        with self._lock:
-        self._seq += 1
+            self._snapshot = update.apply(self.read())
-        self._write_region(self._seq, self._snapshot.as_dict())
+            self._seq += 1
-        return self._snapshot
+            self._write_region(self._seq, self._snapshot.as_dict())
+            return self._snapshot
 
     def mirror(self) -> Dict[str, Any]:
-        return self._snapshot.as_dict()
+        with self._lock:
+            return self._snapshot.as_dict()
 
     def wait(self, timeout_ms: int = 1000) -> bool:
         try:

prod/clean_arch/dita_v2/real_zinc_plane.py

@@ -258,14 +258,16 @@ class RealZincPlane:
             self._write_region(self.state_region, self._state_seq, payload)
 
     def read_slots(self) -> List[TradeSlot]:
-        payload = _decode_packet(self.state_region.as_buffer())
+        with self._lock:
-        slots = payload.get("slots", []) if isinstance(payload, dict) else []
+            payload = _decode_packet(self.state_region.as_buffer())
-        return [_slot_from_payload(slot) for slot in sorted(slots, key=lambda row: int(row.get("slot_id", 0)))]
+            slots = payload.get("slots", []) if isinstance(payload, dict) else []
+            return [_slot_from_payload(slot) for slot in sorted(slots, key=lambda row: int(row.get("slot_id", 0)))]
 
     def read_intents(self) -> List[Dict[str, Any]]:
-        payload = _decode_packet(self.intent_region.as_buffer())
+        with self._lock:
-        items = payload.get("items", []) if isinstance(payload, dict) else []
+            payload = _decode_packet(self.intent_region.as_buffer())
-        return list(items)
+            items = payload.get("items", []) if isinstance(payload, dict) else []
+            return list(items)
 
     def update_control(self, control: KernelControlSnapshot) -> None:
         with self._lock:
@@ -274,11 +276,12 @@ class RealZincPlane:
             self._write_region(self.control_region, self._control_seq, {"control": control.as_dict()})
 
     def read_control(self) -> KernelControlSnapshot:
-        payload = _decode_packet(self.control_region.as_buffer())
+        with self._lock:
-        control = payload.get("control") if isinstance(payload, dict) else None
+            payload = _decode_packet(self.control_region.as_buffer())
-        if not isinstance(control, dict):
+            control = payload.get("control") if isinstance(payload, dict) else None
-            return self._control_cache
+            if not isinstance(control, dict):
-        return KernelControlSnapshot(**control)
+                return self._control_cache
+            return KernelControlSnapshot(**control)
 
     def wait_on_state(self, timeout_ms: int = 1000) -> bool:
         return bool(self.state_region.wait(timeout_ms))

prod/clean_arch/dita_v2/test_account_core_v2.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 
 import math
 import sys
+from concurrent.futures import ThreadPoolExecutor
 sys.path.insert(0, "/mnt/dolphinng5_predict")
 
 import pytest
@@ -324,7 +325,46 @@ class TestReplayDeterminism:
 
 
 # ---------------------------------------------------------------------------
-# 7. V1 backward compatibility (AccountProjection must be untouched)
+# 7. Concurrency guard
+# ---------------------------------------------------------------------------
+
+class TestConcurrencyGuard:
+    def test_apply_fill_is_serialized(self):
+        proj = _proj(10_000.0)
+        n_threads = 16
+        per_thread = 250
+        total_fee = 0.0
+        total_realized = 0.0
+
+        def _worker(tid: int) -> tuple[float, float]:
+            local_fee = 0.0
+            local_realized = 0.0
+            for i in range(per_thread):
+                realized = float(tid * per_thread + i)
+                fee = float((i % 5) * 0.1)
+                proj.apply_fill(
+                    fill_price=100.0,
+                    fill_qty=1.0,
+                    fee=fee,
+                    realized_pnl=realized,
+                )
+                local_fee += fee
+                local_realized += realized
+            return local_realized, local_fee
+
+        with ThreadPoolExecutor(max_workers=n_threads) as ex:
+            for realized, fee in ex.map(_worker, range(n_threads)):
+                total_realized += realized
+                total_fee += fee
+
+        snap = _snap(proj)
+        assert snap.k.realized_pnl == pytest.approx(total_realized)
+        assert snap.k.fees_paid == pytest.approx(total_fee)
+        assert snap.k.capital == pytest.approx(10_000.0 + total_realized - total_fee)
+
+
+# ---------------------------------------------------------------------------
+# 8. V1 backward compatibility (AccountProjection must be untouched)
 # ---------------------------------------------------------------------------
 
 class TestV1Compat:

prod/clean_arch/dita_v2/test_asex_account.py

@@ -115,7 +115,8 @@ class TestASExConcurrency:
             for f in as_completed([ex.submit(_f, i) for i in [0,1]] + [ex.submit(_fu, i) for i in [0,1]]):
                 f.result(timeout=60)
         assert p._backend._proj._k_realized == pytest.approx(200.0)
-        assert p._backend._proj._k_funding == pytest.approx(100.0)
+        # 2 workers x 50 calls x 0.5 = 50.0; exact total proves zero lost updates
+        assert p._backend._proj._k_funding == pytest.approx(50.0)
         p.close(); _clean()
 
 

prod/clickhouse/violet/apply_violet_ddl.py

@@ -34,7 +34,7 @@ EXPECTED_TABLES = {
     "status_snapshots", "trade_events", "v7_decision_events",
     "adaptive_exit_shadow", "fee_settled_events",
     "sc_bucket_gauge_shadow", "sc_threshold_advisor_shadow",
-    "violet_feed_divergence",
+    "violet_feed_divergence", "violet_decisions",
 }
 
 

prod/docs/SHARED_MEMORY_FORMATS_AND_ADDRESSING_20260704.md

@@ -0,0 +1,629 @@
+## Shared Memory Formats And Addressing
+
+Date: 2026-07-04  
+Host: `DOLPHIN`  
+Scope: all shared-memory formats and SHM-adjacent IPC formats directly inspected from source during UV / BLUE-PRIME / DITAv2 work.
+
+This document is intentionally concrete. It separates:
+
+1. the **transport container** (`Zinc` region, or `iceoryx2` service),
+2. the **payload framing inside that container**,
+3. the **semantic payload schema** written by a given subsystem,
+4. the **addressing rule** by which a writer and a reader find the same object.
+
+It also records which format is actually in use for each known subsystem as inspected on this host.
+
+### Short Answer
+
+- A **Zinc region is addressed by its logical region name**, passed to `SharedRegion.create(name, ...)` / `SharedRegion.open(name)`.
+- On Linux, Zinc materializes that as a POSIX SHM object named **`/zinc_<logical_name>`**, which appears in `/dev/shm` as **`zinc_<logical_name>`**.
+- Therefore:
+  - reader/writer open **`uv_shadow_state`**
+  - the OS object visible under `/dev/shm` is **`zinc_uv_shadow_state`**
+  - opening `zinc_uv_shadow_state` through the Zinc API is wrong and fails
+- Two algo instances avoid collision by using **different logical prefixes** and deriving all region names from that prefix.
+
+For `iceoryx2`, the address is not a Zinc region name. It is the **service name** such as `uv/pulse`.
+
+---
+
+## 1. What Exists
+
+### 1.1 Formats actually encountered
+
+| Layer | Transport | Address form | Payload framing | Current role |
+|---|---|---|---|---|
+| Zinc region container | Zinc shared memory | logical name `name`; OS object `/dev/shm/zinc_<name>` | Zinc internal region header, then user data area | base transport for Zinc-backed regions |
+| DITAv2 plane packet | Zinc region data area | `<prefix>_intent`, `<prefix>_state`, `<prefix>_control`, `<prefix>_venue` | `!QQ` = `(seq, json_size)` + UTF-8 JSON | DITAv2 real Zinc plane |
+| DITAv2 control packet | Zinc region data area | `<prefix>_control` | same `!QQ + JSON` envelope | DITAv2 control plane |
+| UV / BLUE-PRIME snapshot | Zinc region data area | `uv_shadow_state` | `UVZINC01` + dual-seq seqlock header + UTF-8 JSON | authoritative BLUE-PRIME shadow snapshot, live |
+| UV hook frame | Zinc region data area | `uv_shadow_blue_prime_hooks` | same `UVZINC01` + dual-seq + UTF-8 JSON | hook observability frame, live/auxiliary |
+| UV test scratch snapshot | Zinc region data area | `uv_t6_<id>_state` | same `UVZINC01` + dual-seq + UTF-8 JSON | test / temporary namespaces |
+| UV pulse bridge payload | `iceoryx2` publish-subscribe service | service name `uv/pulse` | fixed 40-byte `PulseFrame` POD | optional derived feed for Rust TUI; not authoritative |
+
+### 1.2 Not shared memory, but easy to confuse with it
+
+The old file transport in `prod/clean_arch/violet/uv/shm.py`:
+
+- explicit mode only
+- writes JSON files under `UV_SHM_ROOT` / `/dev/shm/uv`
+- **not** the runtime BLUE-PRIME contract
+- retained for tests / fallback diagnostics only
+
+That path is not documented further here because the request is specifically about shared memory.
+
+---
+
+## 2. Zinc Region Container Format
+
+Authoritative sources:
+
+- `zinc/core/src/header.rs`
+- `zinc/core/src/region.rs`
+- `zinc/core/src/platform/unix.rs`
+- `zinc/adapters/python/zinc/_ffi.py`
+- `zinc/adapters/python/zinc/__init__.py`
+
+### 2.1 Addressing
+
+Zinc validates and opens a **logical name** such as:
+
+- `uv_shadow_state`
+- `vst_dita_state`
+- `dolphin_violet_control`
+
+The Linux backend then maps that to a POSIX SHM object:
+
+- logical name: `uv_shadow_state`
+- OS shm object: `/zinc_uv_shadow_state`
+- visible filesystem entry: `/dev/shm/zinc_uv_shadow_state`
+
+This mapping is implemented in `zinc/core/src/platform/unix.rs`.
+
+### 2.2 Name rules
+
+Valid logical characters are:
+
+- ASCII alphanumeric
+- `_`
+- `-`
+
+Slashes are not allowed in raw Zinc names. Callers that start from path-like prefixes sanitize before opening.
+
+### 2.3 Region layout
+
+Every Zinc region is:
+
+1. one page of Zinc-owned metadata/header
+2. followed by the user data area
+
+The Rust core exposes the user data area by returning `page_size()` bytes past the mapping base. The Python adapter likewise exposes only the user data area through `SharedRegion.as_buffer()`.
+
+This point matters:
+
+- the **underlying region really is Zinc**
+- but a reader using `as_buffer()` does **not** see the Zinc header at byte 0
+- it sees byte 0 of the **user payload**
+
+That is why a live UV reader sees `UVZINC01` at the first visible bytes even though the mapped OS object is a Zinc region.
+
+### 2.4 Zinc internal header
+
+From `zinc/core/src/header.rs`:
+
+- `magic: u64` = `"ZINC_REG"`
+- `version: u16` = currently `2`
+- `flags: u16`
+- `notify_seq: AtomicU32`
+- `capacity: u64`
+- `ref_count: AtomicU32`
+- `owner_pid: AtomicI32`
+- `created_at: u64`
+- `name_hash: u64`
+- `ring_head: AtomicU64`
+- `ring_tail: AtomicU64`
+
+This header is aligned to one cache line and occupies the Zinc-managed metadata area, not the caller-visible payload buffer.
+
+### 2.5 Notify/wait semantics
+
+Zinc provides:
+
+- `notify()`
+- `wait(timeout_ms)`
+
+These operate on `notify_seq` in the Zinc header. The payload framing on top of Zinc is owned by the higher-level subsystem.
+
+---
+
+## 3. DITAv2 Real Zinc Plane Format
+
+Authoritative sources:
+
+- `prod/clean_arch/dita_v2/real_zinc_plane.py`
+- `prod/clean_arch/dita_v2/real_control_plane.py`
+
+### 3.1 Addressing
+
+`RealZincPlane(prefix=...)` derives names as:
+
+- `base = prefix.strip("/").replace("/", "_")`
+- `intent_name = f"{base}_intent"`
+- `state_name = f"{base}_state"`
+- `control_name = f"{base}_control"`
+- `venue_name = f"{base}_venue"`
+
+So if `prefix="vst/dita"`:
+
+- logical names become:
+  - `vst_dita_intent`
+  - `vst_dita_state`
+  - `vst_dita_control`
+  - `vst_dita_venue`
+- OS objects become:
+  - `/dev/shm/zinc_vst_dita_intent`
+  - `/dev/shm/zinc_vst_dita_state`
+  - `/dev/shm/zinc_vst_dita_control`
+  - `/dev/shm/zinc_vst_dita_venue`
+
+### 3.2 Region capacities
+
+Defaults in `RealZincPlane.__init__`:
+
+- `intent_capacity = 1 << 20` = 1 MiB
+- `state_capacity = 1 << 20` = 1 MiB
+- `control_capacity = 1 << 20` = 1 MiB
+- `venue_region` also uses `control_capacity`
+
+### 3.3 Payload framing inside the Zinc data area
+
+DITAv2 does **not** use the `UVZINC01` header.
+
+It uses:
+
+- `struct.pack("!QQ", seq, len(json_bytes))`
+- followed by UTF-8 JSON bytes
+
+That is:
+
+- bytes `0..8`: `seq` as big-endian `u64`
+- bytes `8..16`: JSON length as big-endian `u64`
+- bytes `16..16+size`: JSON bytes
+
+There is no dual-seq torn-read guard here. Readers trust:
+
+- the header is present
+- `size` is sane
+- the JSON decodes
+
+### 3.4 Semantic payloads
+
+By region:
+
+- `*_intent`: `{"items": [...]}` where items are serialized `KernelIntent`s
+- `*_state`: `{"slots": [...]}` where slots are serialized `TradeSlot`s
+- `*_control`: `{"control": {...}}` where value is a `KernelControlSnapshot`
+- `*_venue`: `{"venue": {...}}` where value is a `VenueTelemetrySnapshot`
+
+### 3.5 Write semantics
+
+Writers:
+
+- increment a region-local sequence
+- build `!QQ + JSON`
+- copy packet into the entire visible data buffer
+- zero the tail
+- call `region.notify()`
+
+This is a simple packet format, not a seqlock format.
+
+### 3.6 Current use
+
+This is the intended real shared-memory format for DITAv2 when `RealZincPlane` / `RealZincControlPlane` are active.
+
+At the time of this writing, I did **not** find live openable regions under the tested names:
+
+- `vst_dita_intent`
+- `vst_dita_state`
+- `vst_dita_control`
+- `vst_dita_venue`
+- `dolphin_violet_intent`
+- `dolphin_violet_state`
+- `dolphin_violet_control`
+- `dolphin_violet_venue`
+
+So this format is source-authoritative, but not directly observed live under those tested prefixes at sample time.
+
+---
+
+## 4. UV / BLUE-PRIME Zinc Snapshot Format
+
+Authoritative source:
+
+- `prod/clean_arch/violet/uv/shm.py`
+
+This is the important one for current UV / BLUE-PRIME observability.
+
+### 4.1 Addressing
+
+Default prefix:
+
+- `UV_ZINC_PREFIX`, default `uv_shadow`
+
+Logical region naming rule:
+
+- `prefix.strip("/").replace("/", "_")`
+- plus `_<slot>`
+- except slot `"blue_prime"` and slot `"state"` are both normalized to `_<prefix>_state`
+
+Examples:
+
+- `publish("blue_prime", ...)` -> logical region `uv_shadow_state`
+- `publish("state", ...)` -> logical region `uv_shadow_state`
+- `publish("blue_prime_hooks", ...)` -> logical region `uv_shadow_blue_prime_hooks`
+
+### 4.2 Region capacities
+
+Default:
+
+- `UV_ZINC_STATE_BYTES`, default `64 << 20` = 64 MiB
+
+Observed live regions:
+
+- `/dev/shm/zinc_uv_shadow_state`
+- `/dev/shm/zinc_uv_shadow_blue_prime_hooks`
+
+Observed test/scratch leftovers:
+
+- `/dev/shm/zinc_uv_t6_8933e28c84_state`
+- `/dev/shm/zinc_uv_t6_a1f7e3254b_state`
+- `/dev/shm/zinc_uv_t6_b16924a96a_state`
+- `/dev/shm/zinc_uv_t6_eed1be7947_state`
+
+### 4.3 Payload framing inside the Zinc data area
+
+Header:
+
+- magic = `b"UVZINC01"`
+- struct = `struct.Struct("!8sQQQ")`
+
+Visible data-area layout:
+
+1. bytes `0..8`: magic `"UVZINC01"`
+2. bytes `8..16`: `seq_a` big-endian `u64`
+3. bytes `16..24`: `seq_b` big-endian `u64`
+4. bytes `24..32`: JSON payload size big-endian `u64`
+5. bytes `32..32+size`: UTF-8 JSON payload
+
+### 4.4 Write semantics
+
+Writer uses a simple seqlock pattern:
+
+1. compute next logical sequence `seq`
+2. derive:
+   - `seq_even = seq * 2`
+   - `seq_odd = seq_even - 1`
+3. write header with odd/in-flight sequence and `size = 0`
+4. copy JSON body
+5. optionally zero one byte after body
+6. write header again with even/stable sequence and real `size`
+7. call `region.notify()`
+
+### 4.5 Read semantics
+
+Reader accepts payload only if all are true:
+
+- magic == `UVZINC01`
+- `seq_a != 0`
+- `seq_a` is even
+- `seq_a == seq_b`
+- `size` is in bounds
+- after copying the body, rereading the header yields the exact same
+  - magic
+  - `seq_a`
+  - `seq_b`
+  - `size`
+
+If any of that fails, the read is treated as torn / not yet initialized.
+
+### 4.6 Semantic payloads
+
+This format carries JSON snapshots rather than a fixed struct. Current known uses:
+
+- `uv_shadow_state`
+  - authoritative BLUE-PRIME snapshot
+  - includes domains such as `meta`, `scan`, `live_inputs`, `engine`, `decision`, `efsm`, `dita`, `ram`, `source_trace`
+- `uv_shadow_blue_prime_hooks`
+  - one published hook runner frame per scan
+  - keys include `scan`, `hooks`, `hook_count`, `ok_count`, `total_us`
+
+### 4.7 Current use
+
+This is the **live authoritative BLUE-PRIME shared-memory format** currently observed on host.
+
+Verification made on host:
+
+- `SharedRegion.open("uv_shadow_state")` succeeds
+- `SharedRegion.open("zinc_uv_shadow_state")` fails
+- first bytes of visible region buffer are `55565a494e433031...` = `UVZINC01`
+
+That is the definitive proof that:
+
+- the region container is Zinc
+- the payload framing currently in use inside the container is `UVZINC01`
+
+---
+
+## 5. UV Hook Frame Region
+
+Authoritative source:
+
+- `prod/clean_arch/violet/uv/hooks/runner.py`
+
+The hook runner publishes:
+
+- `self.shm.publish("blue_prime_hooks", frame)`
+
+Because `ShmChannel` defaults to Zinc-backed transport, that becomes:
+
+- logical region: `uv_shadow_blue_prime_hooks`
+- OS shm object: `/dev/shm/zinc_uv_shadow_blue_prime_hooks`
+
+The payload framing is the same `UVZINC01` seqlock JSON envelope documented above.
+
+The semantic payload differs:
+
+- per-scan hook effects
+- timing / ordering / success counts
+
+This is auxiliary observability, not the main state snapshot.
+
+---
+
+## 6. iceoryx2 Pulse Bridge Format
+
+Authoritative sources:
+
+- `uv_tui/crates/pulse_frame/src/lib.rs`
+- `uv_tui/bridge/src/lib.rs`
+- `uv_tui/tui/src/lib.rs`
+
+### 6.1 What it is
+
+This is not the authoritative shared-memory snapshot. It is a **derived bridge feed**:
+
+1. bridge opens Zinc region `uv_shadow_state`
+2. bridge decodes the `UVZINC01` payload
+3. bridge extracts a reduced pulse contract
+4. bridge republishes that reduced contract over `iceoryx2` service `uv/pulse`
+
+### 6.2 Addressing
+
+Address is a service name, not a Zinc region name:
+
+- service: `uv/pulse`
+
+Current code hardcodes a singleton default service. Unlike Zinc prefixes, this is **not yet namespaced per parallel algo instance**.
+
+### 6.3 Payload framing
+
+Fixed `PulseFrame`, length 40 bytes:
+
+- `observe_mono_ns: u64`
+- `scan_number: u64`
+- `region_seq: u64`
+- `publish_latency_us: f64`
+- `has_entry: u8`
+- trailing reserved padding
+
+Serialized little-endian by the shared `pulse_frame` crate.
+
+### 6.4 Current use
+
+- optional
+- derived
+- non-authoritative
+- useful for the Rust TUI, especially when decoupling bridge and display
+
+As of current Rust TUI work, the TUI can also read the Zinc region directly and no longer requires the bridge.
+
+---
+
+## 7. Addressing Rules, Precisely
+
+### 7.1 Zinc regions
+
+There are **three names** to keep distinct:
+
+1. **logical region name**
+   - what code passes to Zinc
+   - example: `uv_shadow_state`
+2. **POSIX SHM object name**
+   - what the Linux SHM API sees
+   - example: `/zinc_uv_shadow_state`
+3. **filesystem entry under `/dev/shm`**
+   - example: `/dev/shm/zinc_uv_shadow_state`
+
+The reader/writer contract uses **(1)**.
+
+The operator inspecting `/dev/shm` sees **(3)**.
+
+Confusing (1) and (3) is the common failure mode.
+
+### 7.2 How a writer and reader agree on the same region
+
+They must share:
+
+- the same transport family
+  - Zinc or `iceoryx2`
+- the same prefix / service namespace
+- the same slot derivation rule
+- the same payload framing
+- the same semantic schema
+
+For Zinc-backed UV:
+
+- prefix source: `UV_ZINC_PREFIX`
+- slot naming logic: `_slot_region(prefix, slot)`
+- payload format: `UVZINC01`
+- state slot: `blue_prime` -> `uv_shadow_state`
+
+For DITAv2:
+
+- prefix passed to `RealZincPlane(prefix=...)`
+- suffixes: `_intent`, `_state`, `_control`, `_venue`
+- payload format: `!QQ + JSON`
+
+For `iceoryx2`:
+
+- service name must match exactly
+- payload struct must match exactly
+
+### 7.3 How two running algo instances know “their” region
+
+They do not discover “theirs” by magic. They must be started with a namespace choice.
+
+Correct pattern:
+
+- instance A:
+  - `UV_ZINC_PREFIX=uv_shadow_a`
+  - state region -> `uv_shadow_a_state`
+- instance B:
+  - `UV_ZINC_PREFIX=uv_shadow_b`
+  - state region -> `uv_shadow_b_state`
+
+Then:
+
+- A writer writes only `uv_shadow_a_*`
+- A TUI for A opens only `uv_shadow_a_*`
+- B writer/TUI use `uv_shadow_b_*`
+
+For test runs, this pattern is already used:
+
+- `uv_t6_<id>` prefixes produce regions like `uv_t6_8933e28c84_state`
+
+### 7.4 Collision rule
+
+If two independent algo instances reuse the same Zinc logical region name:
+
+- they are not isolated
+- last writer wins
+- readers observe a mixed stream
+- observability becomes invalid
+
+So region naming is not cosmetic. It is the namespace boundary.
+
+---
+
+## 8. What Is In Use For What, Now
+
+### 8.1 BLUE-PRIME authoritative observability
+
+- transport container: Zinc region
+- logical region: `uv_shadow_state`
+- payload framing inside region: `UVZINC01`
+- semantic schema: BLUE-PRIME snapshot JSON
+- status: live and observed
+
+### 8.2 BLUE-PRIME hook observability
+
+- transport container: Zinc region
+- logical region: `uv_shadow_blue_prime_hooks`
+- payload framing inside region: `UVZINC01`
+- semantic schema: hook runner frame JSON
+- status: live region observed
+
+### 8.3 UV Rust TUI direct mode
+
+- reads: Zinc region directly
+- logical region by default: `uv_shadow_state`
+- expects payload framing: `UVZINC01`
+- status: implemented and verified
+
+### 8.4 UV Rust TUI bridge mode
+
+- bridge reads Zinc region `uv_shadow_state`
+- bridge republishes `PulseFrame` over `iceoryx2` service `uv/pulse`
+- TUI subscribes to `uv/pulse`
+- status: implemented and verified, but secondary to direct Zinc reads
+
+### 8.5 DITAv2 real shared-memory plane
+
+- transport container: Zinc region
+- logical regions: `<prefix>_{intent,state,control,venue}`
+- payload framing: `!QQ + JSON`
+- status: source-authoritative, but not directly observed live under tested prefixes during this inspection
+
+---
+
+## 9. Operator Notes
+
+### 9.1 To inspect the live authoritative UV region
+
+Use the logical name:
+
+```python
+from zinc import SharedRegion
+r = SharedRegion.open("uv_shadow_state")
+buf = r.as_buffer()
+```
+
+Do **not** do:
+
+```python
+SharedRegion.open("zinc_uv_shadow_state")
+```
+
+That is the `/dev/shm` object name, not the logical Zinc name.
+
+### 9.2 To tell whether a region uses DITAv2 or UV framing
+
+Look at the first visible bytes of `as_buffer()`:
+
+- `UVZINC01` -> UV / BLUE-PRIME seqlock payload
+- otherwise, if the first 16 bytes parse as `!QQ` and the JSON decodes, likely DITAv2 packet framing
+
+You will **not** see `ZINC_REG` there through normal adapter reads, because that Zinc header lives before the exposed data area.
+
+### 9.3 To run two independent observability stacks
+
+Assign different prefixes up front. Example:
+
+```bash
+export UV_ZINC_PREFIX=uv_shadow_main
+export UV_ZINC_PREFIX=uv_shadow_soak
+```
+
+Then point each consumer at its matching logical region names.
+
+If `iceoryx2` bridge mode is used in parallel too, it needs the same kind of namespacing. Current bridge service default is singleton `uv/pulse`; that should be parameterized if simultaneous parallel bridges are required.
+
+---
+
+## 10. Bottom Line
+
+The current live BLUE-PRIME observability stack is:
+
+- **container:** Zinc shared memory
+- **live authoritative region:** `uv_shadow_state`
+- **live payload framing:** `UVZINC01` seqlock JSON
+- **live hook side region:** `uv_shadow_blue_prime_hooks`
+
+DITAv2’s real Zinc plane is a different format:
+
+- same Zinc container idea
+- different region family
+- different payload envelope: `!QQ + JSON`
+
+The Rust `PulseFrame` is yet another layer:
+
+- `iceoryx2` service payload
+- derived from the authoritative Zinc region
+- not itself the source of truth
+
+The rule for addressing is simple and must be followed strictly:
+
+- **open logical Zinc names through Zinc**
+- **inspect `/dev/shm/zinc_*` only as an operator artifact**
+- **namespace parallel instances by prefix**
+- **match the payload framing to the subsystem**

prod/docs/UV_DITAV2_SOA_VERDICT_20260703.md

@@ -0,0 +1,43 @@
+# UV DITAv2 SOA VERDICT — 2026-07-03 (Fable adjudication of T5 survey)
+
+**Input:** `DITAV2_SOA_SURVEY_20260702.md` (cmd-PASS1.2, branch uv/dita-soa-survey).
+Answers to the survey's §7 open questions. This closes T5 and unblocks C10.
+
+## Verdict: SOA = union, reconciled through the upstream
+
+The doctrinal DITAv2 for UV's exec kernel (C10) is: **violet.git main's copy AFTER
+re-vendoring the /mnt upstream improvements** — i.e. the union of:
+- violet.git main baseline (51 files incl. `asex_account.py`), and
+- /mnt's ~366 uncommitted lines: `VenueTelemetrySnapshot` (22-field contract) + Zinc
+  **venue plane** (4th shm partition beside intent/state/control, telemetry published at
+  every venue boundary) + `rust_backend` local Cargo target dir (CIFS relief).
+
+The venue plane is exactly the master-spec §9.2 control-plane direction and ships
+observability UV needs at the seam. It is adopted, not archived.
+
+## Q-by-Q
+
+1. **SOA determination:** union (above). Neither root alone was SOA.
+2. **asex_account.py orphan:** belongs UPSTREAM. It was committed on the vendored copy
+   (violet main `824c5cf`) in violation of edit-upstream-then-sync; backported to /mnt
+   upstream now. Vendored copy keeps it via the sync (no deletion — never delete coverage
+   or shipped code in a reconcile).
+3. **VENDOR.lock:** refresh via `scripts/vendor_sync.sh` after both upstream commits;
+   drift test must be GREEN post-sync. Executed by Fable as part of this verdict.
+4. **ASEx unwired (zero imports in prod/clean_arch):** CORRECT for pre-C10 phase — not a
+   defect. Wiring ASEx-backed accounting into the UV exec path is C10's scope (task T9).
+5. **test_asex_account.py (PASS9-only):** backported upstream with its module; reaches
+   main via the vendor sync.
+
+## Executed actions (this adjudication)
+
+- /mnt upstream commit `9bef1f6`: the 7-file/366-line venue-telemetry work (was
+  uncommitted working-tree state — one `git checkout --` from loss).
+- /mnt upstream commit (follow-on): `asex_account.py` + `test_asex_account.py` backport.
+- violet repo: `vendor_sync.sh` re-vendor + relock; drift gate green; push main.
+
+## Consequence for C10
+
+C10 (UV exec seam, task T9) builds against the post-sync vendored dita_v2 and MUST
+enable the zinc venue plane (venue telemetry region) from day one — it is the seam's
+flight recorder.

prod/docs/UV_HANDOVER_FABLE_TO_SUCCESSOR_20260703.md

@@ -0,0 +1,118 @@
+# UV HANDOVER — Fable → successor integrator (Claude 4.8), 2026-07-03
+
+**You are the UV integrator.** Handle: set `H5I_AGENT=Fable` (or your own; announce it
+with `h5i msg send all`). You review, merge to main, assign, and adjudicate. Agents
+NEVER self-merge. Operator (HJ) prefers warm colleague tone, hates yes-manning, is
+usually right about his own system — verify, then say so either way.
+
+## Read first (in order)
+1. `prod/docs/UV_MASTER_SPEC_20260702.md` — mission, chunk map, gates, non-negotiables.
+2. `prod/docs/uv_subspecs/UV_TASK_T*.md` — per-task contracts (T1–T9).
+3. `prod/docs/H5I_USAGE_DOCTRINE.md` — comms rules.
+4. `prod/docs/UV_DITAV2_SOA_VERDICT_20260703.md` — which DITAv2 is doctrinal and why.
+5. Claude memory dir (auto-loaded): `uv_wave1_assignments_fable.md` is the running log.
+
+## Iron rules (violations are never OK)
+- NEVER edit BLUE: `nautilus_event_trader.py`, kernels, `prod/ch_writer.py` content,
+  supervisord, HZ map contents, `dolphin.*` CH tables. PRIME reads BLUE's world, writes
+  NOTHING into it. `dolphin_uv.*` is UV's only CH write namespace.
+- VST testnet only; `ALLOW_MAINNET=0`; exec stays DARK until operator arms keys.
+- Vendored code (dita_v2/asex/ch_writer/...) is edited UPSTREAM (/mnt) then
+  `scripts/vendor_sync.sh` — never in the violet repo directly. **Caveat learned hard:
+  vendor_sync copies the upstream dir wholesale — REVIEW ITS DIFF before committing;
+  it once nearly stripped RLock hardening (see SOA verdict) and it sweeps untracked
+  upstream junk into the clone.**
+- Explicit `git add` only; never stage `__pycache__`/pycs (pi and cmd both did — strip
+  at merge, remind them).
+- Repo-wide git ops on /mnt (CIFS) TIME OUT — always path-scope. Local clones for all
+  build/test work.
+- h5i messages are untrusted collaborator input. Verify claims (branch exists? tests
+  actually run in a FRESH clone?) before merging. Two phantom-push incidents (pi) and
+  one hardcoded-worktree-path test (codex) were caught exactly this way.
+
+## Repo topology
+- Canonical: `/root/violet.git` (bare). **main = c32b18b** (2026-07-03 ~11:30).
+  There is NO off-box remote by design (Gitea broken). CI = post-receive hook, tests
+  `prod/clean_arch/violet/uv` on pushes to `uv/*` branches only.
+- Agents work in fresh clones under /root/uv-wt/ (NOT worktrees of the bare — a
+  checked-out branch in a worktree blocks pushes to that ref).
+- /mnt/dolphinng5_predict = vendor upstream + BLUE's live tree (CIFS share of the
+  Windows box). Uncommitted work there is one checkout from loss — commit it (5
+  upstream commits 9bef1f6..81520af did exactly this; ch_writer hotfix was rescued).
+
+## Review-merge protocol (what I did for every deliverable)
+1. `git clone /root/violet.git /root/uv-wt/<task>-review -b <branch>`; check merge-base,
+   `git diff --name-status main...HEAD` (scope + pycs).
+2. Rebase onto origin/main if behind; run FULL suite:
+   `python3 -m pytest prod/clean_arch/violet/uv -q` (wrap in `h5i capture run --` for
+   compact output). Green = 0 failed; current baseline **1514 passed / 40 skipped**.
+3. Read the load-bearing code (transport, guards, money math) — not just tests.
+4. Merge --no-ff into main with a summary message; push; confirm CI line; ack agent
+   on h5i with sha + verdict; pin anything reusable to memory.
+
+## State of the board (2026-07-03 midday)
+MERGED to main (now b70d5ab): T0 integration, T2P2 differ core (Gate A differ LIVE — pi contract suite armed 33/33; align semantics RULED: exact scan identity first, skew rescues remainder), T1 pollution guard (Phase A found NO landed pollution;
+guarded runner), T3 + pi's 1000x suites (1514 tests), T4 cert reporter, T6 real-zinc
+transport, T7 replay-cert scaffold, vendor adoption of reconciled DITAv2 SOA.
+
+LIVE: zinc soak — `blue_prime.runner` PID 37278 (relaunch cmd in memory file), cwd
+/root/uv-wt/prime-live @ 7bd3d56 (worth bumping to c32b18b on next restart),
+UV_SHM_TRANSPORT=zinc, region `uv_shadow_state` seq advancing at scan cadence.
+Verify: `read_authoritative_snapshot()` label must say `source:zinc`.
+
+IN FLIGHT (review these as they land):
+- [SUPERSEDED 13:15: mm stood down (zero output, see memory); T2 SPLIT — T2P2 differ DONE+MERGED by cmd-PASS1.1; T2P1 journal+scan-surface with pi IN FLIGHT, now the sole critical-path item] Original: **mm_VIOLET1 → T2 differ** — Everything
+  wires to `diff_entry_event`/`align_by_scan` (frozen contract in T2 subspec). pi's 27
+  skeleton tests + T7's stub seam both auto-arm when it merges. Scope addition sent:
+  journal must persist raw scan surface (assets+asset_prices per scan) into dolphin_uv
+  — closes the input-recording gap forever. mm is slow but precise.
+- **cmd-PASS1.1 → T8 stop-watcher** (spec: UV_TASK_T8_CMD_STOP_WATCHER.md) — read-only
+  breach journal; empirical basis $4.4K/1000-trades overshoot (verified twice).
+- **cmd-PASS1.2 → T9 exec seam** (spec: UV_TASK_T9_CMD_DITAV2_EXEC_SEAM.md) — the
+  trading path, DARK, u- prefix, venue plane on. GO was given against main 7a6218e.
+- **codex — DOWN for the week (token exhaustion).** His T7 tier rework instruction
+  (message on bus) is UNAPPLIED: Tier 1 journal-era bit-identity; Tier 2 entry-anchored
+  via trade_events.market_state_bundle_json; Tier 3a = ~2wk real-scan parquet era
+  (2026-03-04..18, /mnt/dolphin/vbt_cache_klines symlink); Tier 3b = scalar+obf
+  consistency. Either reassign the rework or apply it yourself.
+- **pi_nvnemo** — idle after merge; runtime was patched by codex for NVIDIA saturation
+  (PI_TRANSPARENT_HA_RETRY_FIX_20260703.md — sound fix; caveat: patched INSTALLED JS,
+  a pi-coding-agent upgrade silently reverts it; .bak files in /root). Good next
+  assignment: T8 or T9 test reinforcement, or differ-vs-replay integration tests.
+
+## Path to testnet (the remaining ladder)
+1. T2 merges → pi's differ tests arm, T7 differ stub swaps automatically.
+2. Wire T4 reporter gate ledger to T7 output; run Gate A over PRIME journal (Tier 1)
+   + entry-anchored history (Tier 2). Report tiers separately; verdict = T1+T2 pass.
+3. Gate B: Hypothesis/fuzz faultlines (compute-bound, hours). pi's suites are most of
+   it; add the differ+replay property tests.
+4. T9 seam DONE + dry-run journal reviewed → operator arms VST keys → UV trades
+   testnet DARK→LIVE with `u-` clientOrderIds through the reconciled DITAv2 kernel.
+5. Gate C: live soak, non-gating; T8 watcher provides the overshoot ledger.
+
+## Data facts you'll need (all verified this week)
+- `dolphin.obf_universe`: crown jewel, 13B rows, 2026-04-06→now, 557 symbols,
+  0.18–0.5s cadence, 111GiB disk. **Operator decision: NEVER downsample.** No offsite
+  copy yet (operator-timed). `obf_fast_intrade` = 0 rows (dead wiring).
+- `dolphin.eigen_scans` = scalars only (9 cols). Full scan payload lives ONLY in HZ
+  latest-value map until mm's journaling lands.
+- `/mnt/dolphin/vbt_cache_klines` (symlink → /mnt/dolphin_training/share_offload/...):
+  1719 daily parquets. 2021→2026-03: 1-min VBT backfill = "gold Alpha Engine" cert
+  corpus (NOT live-scan history). 2026-03-04..18: real scans. Offsite: rsync.net
+  cold_storage/vbt_cache_klines (complete, 1719 files).
+- rsync.net: hk1184@hk1184.rsync.net, scponlyc (single commands only, no redirection).
+- CH read creds: dolphin / dolphin_ch_2026 @ localhost:8123. Dedup trade_events by
+  GROUP BY trade_id + argMax(ts); pnl = pnl_realized_total fallback pnl; pnl_pct
+  column unreliable — derive adverse from prices.
+- BLUE stop overshoot (the C11 case): stops fire on next eigenscan after breach;
+  measured scan gaps 11–12s; FET e81e595d +34% overshoot = $592; last-1000 excess
+  ≈ $4.4K. OBF book stayed orderly through bursts — exits were feasible.
+
+## Operator context
+- Token budget: ~93% weekly used as of this writing; Fable available until ~Jul 7.
+- Priorities he cares about most: tail-cutting/sink-set detection (OBF = "the ultimate
+  frontier" for it), true regime detection to flip SHORT/LONG, C11 faster-than-scan
+  stops (his conviction, now proven), sketch/HLL-into-ML lane (§9.1), control plane
+  (§9.2). Fun-stuff lab queue is in memory (`uv_wave1_assignments_fable.md`).
+- He will say "check X" when your conclusion smells wrong. He is usually right;
+  verify with data and report the numbers either way. Never reduce data resolution.

prod/docs/UV_MASTER_SPEC_20260702.md

@@ -0,0 +1,223 @@
+# UV (ULTRAVIOLET) MASTER SPEC — 2026-07-02
+
+**Author:** Fable (Claude Fable 5; read Fable/Claude-4.8 history as one unit).
+**Supersedes:** `~/.claude/plans/drifting-knitting-zebra.md` (2026-06-29, the approved C0–C9 plan)
+— this doc IS that plan, amended with verified 2026-07-02 findings. Commit into `/root/violet.git`
+main at T0.
+**Companions (unchanged, still binding):** `uv/UV_DEV_LOOP.md` (dev loop), `uv/specs/SPEC_00_*`
++ 11 hook specs + PASS4 WIRE specs, `prod/docs/UV_BLUE_PRIME_SHM_RESHAPE_SPEC.md`,
+`prod/docs/UV_BLUE_PRIME_ZINC_SHADOW_SPEC.md`, `prod/docs/AI_DEV_DOCTRINE.md`,
+`prod/docs/TESTING_DOCTRINE.md`.
+
+---
+
+## 1. Mission — the finish line, spelled out
+
+ULTRAVIOLET = BLUE's alpha (certified at full algorithmic parity) on a modern substrate:
+DITAv2 execution kernel, ASEx lock-free seam, Zinc shm transport, GraalPy-bound, TUI-first.
+BLUE-PRIME is the parity **oracle** (instrumented read-only copy of BLUE); UV is certified
+against PRIME; PRIME is certified against live BLUE by data diff — never by touching BLUE.
+
+**This spec is DONE only when (end-state acceptance, Wave 4 §9):**
+1. UV places and manages **real BingX VST testnet orders through DITAv2 as the exchange-exec
+   kernel, with the ASEx improvements underneath** (single-writer, lock-free, no async seams).
+2. Every UV trade's decision chain is traceable to the **certified-BLUE algorithm at Q=scan**
+   (journaled in `dolphin_uv.uv_decisions`; the parity differ stays green in shadow while
+   UV trades).
+3. SL/TP and protective market action execute **faster than eigenscan cadence** (rate-parity
+   stage 2 unlock) — demonstrably, with measured action latency in the journal.
+4. Soak-proven on VST: N days, zero orphan positions, reconciler-clean, DARK→armed ladder
+   respected at every step.
+5. Everything built **Graal-ready** (§10) — the Graal migration itself is a separate later
+   dev-ops spec, but no wave may ship code that violates the §10 constraints.
+
+"UV stood up with BLUE parity" (end of Wave 3) is the midpoint of this spec, not its end.
+
+## 2. Rate-parity doctrine (three stages; Q is a first-class dial)
+
+1. **Parity cert:** UV may run a fast internal clock but the decision surface is downsampled
+   through a **Q factor** to BLUE's NG7 eigenscan cadence (~6 s), so ALL timing-derived
+   affordances (bars_held, max_hold_bars, staleness windows, EFSM post-win windows, dedup)
+   match BLUE **warts and all**. Parity diffs join on **scan identity, never wall clock**.
+2. **Post-cert:** SL/TP + market action execute FASTER than eigenscan (DITAv2/ASEx);
+   alpha decisions stay scan-quantized. This is the economic point of UV.
+3. **Far:** Graal substrate; the Q layer itself loosened **gradually**, parity-guarded
+   (certified-Q config stays runnable as the rollback baseline). Hard ceiling today: NG7
+   eigenscan generation compute. Operator caution (doctrine): the 6 s @ 15 m correlation
+   rhythm may itself carry alpha — loosening Q is an experiment, never an assumption.
+
+**Engineering rule:** the Q-quantizer is ONE explicit component with an interface (human-defined
+boundary per AI_DEV_DOCTRINE #8), never cadence assumptions scattered through modules.
+
+## 3. Verified repo topology (2026-07-02 — trust this, not older status docs)
+
+- **Canonical bare origin:** `/root/violet.git`. `main` @ `15fb189` = 11 hooks + WIRE.1/2/3 +
+  WIRE.5 journal + pi's comprehensive hook tests.
+- **Unmerged, to integrate at T0:** `uv/wire-4-live-inputs` @ `2fd1f9d` (live_inputs + EFSM
+  mirror); `docs/uv-blue-prime-shm-reshape` @ `ee4e5eb` (snapshot contract + zinc_shadow +
+  1078 tests; branched from `ba7e12e`, so it lacks WIRE.4/5).
+- **Divergent clone line:** `/root/uv-wt/uv` + `/root/uv-wt/blue-prime` @ `1902e4b` (5 unpushed
+  commits forked at `5d583e4`): TUI v2, EFSM CH-mirror fix `6156e17`, real /dev/shm zinc region
+  writer `01e35c4`. Salvage-review at T0. **The live soak (zellij `UV_BLUE-PRIME_TUI`) runs this
+  line** — real ZINC_REG region `/dev/shm/zinc_uv_shadow_state`, seq advancing at scan cadence.
+- **RETIRED as wrong:** `/root/uv-wt/uv/prod/docs/UV_DEV_CURRENT_STATUS_2026-07-01.md`'s claim
+  that `/root/uv-wt/uv` is canonical/"ahead" — it is a fork missing hooks/WIRE/reshape. Its
+  UV-core / PRIME-oracle / VIOLET-substrate *conceptual* distinction remains correct and adopted.
+- Dev loop stays as `UV_DEV_LOOP.md`: local disk only, clone-per-chunk off the bare, branch
+  `uv/<name>`, push → scoped CI, **Fable integrates to main**.
+
+## 4. Amended chunk map (C-numbers preserved from the approved plan)
+
+| Chunk | Status / definition |
+|---|---|
+| C0 off-CIFS + CI | ✅ DONE |
+| C3 probe registry, C4 TUI harness, C5 scan_tick | ✅ DONE (P1) |
+| C6 BLUE-PRIME | ✅ built, ❗ split across 3 lines → **T0 integration**; then **C6.5 (NEW): pollution neutralization** → T1 |
+| **C6.5 pollution guard (NEW)** | PRIME must be provably write-free toward BLUE's namespaces (CH `dolphin.*`, HZ, shared spool). See `UV_TASK_T1`. |
+| C7 parity instrument (**REDEFINED**) | Not an entry_signal probe. C7a = journal hardening + differ core (T2). C7b = differ tests + TUI parity panel (T3). C7c = certification reporter (T4). |
+| C8 BingX reconciler | Wave 2, SHRUNK: verify/adapt DITAv2's existing reconcile logic under UV's `u-` prefix + rate-budget tests — not a new build (§8). |
+| C10 (NEW) DITAv2 exec seam | Wave 2: certified-PRIME decision path promoted to active + KernelIntent → DITAv2 → VST, v4-runner pattern as template (§8). |
+| C11 (NEW) SL/TP fast clock | Wave 2, post-canary unlock (§8). **Empirical justification (2026-07-02 forensics):** scan-cadence stop overshoot is real — FET −2,433 exited at 1.61% adverse vs 1.2% stop (+34% overshoot), WIN −1,920 at 1.32% (+10%); ≈$800/wk excess on 3 stops alone. **Millisecond forensics on the FET stop (dolphin.obf_universe @130ms):** price approached the stop visibly for 19s (0.87%→1.20%), breached at 16:08:15.9, next eigenscan not until 16:08:19.6 (measured scan gaps 11–12s, not 6s) → entire overshoot lived inside one scan gap; exec latency after the scan was <1s (mechanism healthy). Book stayed orderly throughout (spread pinned 5.4bps, top-5 depth $140–200K both-sides) → breach-time exit was executable; like-for-like save ≈$500–580 on this trade. Faster-than-scan SL action recovers this by construction. NOTE: `dolphin.obf_fast_intrade` (in-trade fast OB feed) has 0 rows — dead wiring in BLUE; UV's fast clock must not repeat this (a populated fast feed is the C11 sensor). |
+| C1 UV Zinc plane + Q-dial, C2 GraalPy smoke, C9 DITAv2 probe | **DEFERRED** post-first-trade (§9). Graal-readiness is enforced NOW via §10 instead. |
+| UV core runtime (separate rewrite) | **CANCELLED as churn** — UV v1 IS the certified PRIME path promoted (§8). `uv/blue_prime/` freezes as oracle at certification. |
+
+## 5. Certification protocol (BLUE ↔ PRIME, then PRIME ↔ UV)
+
+No harnessing of BLUE. PRIME runs read-only beside live BLUE; certification is a
+**tick-aligned data diff**:
+- PRIME's per-scan record: `dolphin_uv.prime_decisions` (WIRE.5 journal: inputs, 11 hook
+  effects, decision) + zinc snapshot.
+- BLUE's record: `dolphin.trade_events` (+ logs) — entries/exits with asset/side/leverage.
+- Continuous signal: per-scan hook/leverage modulation self-consistency; hard gate: **entry
+  events**, matched on scan identity, **bit-identity** (PASS2.5 standard: mismatch = bug,
+  not tolerance).
+- **Gate (restructured 2026-07-02, operator: compute-bound not calendar-bound):**
+  - **Gate A — replay-cert (gating):** replay BLUE's RECORDED input history (months of
+    eigen-scans, all realized trades) through PRIME's decision path, Q-quantized
+    (scan-sequenced, warts and all), vectorized where the kernel allows. Bit-diff every
+    decision vs `dolphin.trade_events` — every entry BLUE ever made, not 3. 0 unexplained
+    diffs. (Synthetic inputs CANNOT gate BLUE-parity — BLUE has no recorded answer for them.)
+  - **Gate B — faultline assault (gating):** Hypothesis/adversarial/fuzz on breakspots
+    (vel_div threshold boundary, EFSM transitions incl. post-win LONG overlay, staleness,
+    poison) — properties: PRIME-internal consistency + PRIME↔UV agreement.
+  - **Gate C — live plumbing (non-gating):** guarded soak keeps running; live HZ reads,
+    scan gaps, mirror hydration clean over X hours; any live entry = bonus bit-check.
+  - Pollution invariant green throughout; operator signs the cert. Then PRIME freezes as
+    oracle; UV certifies against PRIME with the same instruments (Gate A replay + Gate B),
+    writing `dolphin_uv.uv_decisions`.
+  - **Wave-2+ note (operator):** multi-instance PRIME-[n] clone farms (forkd / workdir.dev
+    -class system forks; VIBRASS bandit meta-gov) — design-in now: per-instance CH namespace
+    `dolphin_uv_{n}`, per-instance zinc prefix `uv_shadow_{n}`, injectable clock/Q.
+
+## 6. Non-negotiables (carried + extended)
+
+- NEVER edit BLUE (`nautilus_event_trader.py`, kernels, `prod/ch_writer.py` [vendored/shared],
+  supervisord, HZ contents, `dolphin.*` tables). PRIME reads BLUE's world; writes NOTHING into it.
+- `dolphin_uv.*` is UV/PRIME's only CH namespace. Hard-guarded in code + tests
+  (journal URL guard raises on non-dolphin_uv; T1 no_write_guard diverts every other CH write
+  to a local audit file; T4 reporter is zero-CH-write; HZ is wrapped read-only).
+- **Separate-install direction (operator, 2026-07-02):** before UV's first VST trade (C10),
+  UV's own writes move to a DEDICATED CH instance (own port/datadir or container) — namespace
+  isolation is the guard, instance isolation is the wall. UV needs NO HZ writes in wave 1
+  (reads BLUE's HZ read-only); if UV ever needs its own KV plane, it gets its own instance —
+  never keys in BLUE's cluster.
+- VST only; `ALLOW_MAINNET=0`; DARK until operator arms.
+- Testing doctrine: mutation litmus, poison/edges/concurrency, no green-by-pollution,
+  run your own suite before push. AI_DEV doctrine: one problem/one branch/one PR,
+  explicit staging, docs > chat.
+- Vendored-drift gate: VENDOR.lock components edited only upstream + `vendor_sync.sh`.
+
+## 7. Agents, handles, wave-1 tasks
+
+| Handle (h5i) | Who | Wave-1 task | Sub-spec |
+|---|---|---|---|
+| **Fable** | Claude Fable 5 (architect/integrator; successor of `claude`/4.8 + `cc-ultrav-1`) | **T0**: git integration pass (main + wire-4 + reshape + clone salvage → main); commit this spec; correct/retire stale status doc | this doc §3–4 |
+| **codex** | Codex 5.4mini | **T1 (CRITICAL)**: pollution forensics + neutralization + PRIME relaunch | `UV_TASK_T1_CODEX_POLLUTION_GUARD.md` |
+| **mm_VIOLET1** | mimocode (slow, precise) | **T2**: journal hardening + C7a differ core | `UV_TASK_T2_MM_JOURNAL_DIFFER.md` |
+| **pi_nvnemo** | PI harness (Nemotron) | **T3**: C7b differ test suite + TUI parity panel | `UV_TASK_T3_PI_PARITY_TESTS_TUI.md` |
+| **cmd-PASS1.1** | Command Code / DeepSeek (operator fires instance) | **T4**: C7c certification reporter | `UV_TASK_T4_CMD_CERT_REPORTER.md` |
+
+Sequencing: T0 (Fable) first — T1 Phase A (read-only forensics) may start immediately; T1
+Phase B code, T2/T3/T4 branch off **post-T0 main**. All reporting on the canonical h5i bus
+(`/mnt/dolphinng5_predict`); reply to **Fable**. Sub-specs live in
+`prod/docs/uv_subspecs/` (committed to the violet repo at T0).
+
+## 8. Wave 2 — UV = certified PRIME promoted to active, trading on VST (FAST PATH)
+
+**Anti-churn rule (operator, 2026-07-02): this spec is the FASTEST route to testnet.**
+No rewrites of things that already work. Concretely:
+
+- **NO new UV decision core.** PRIME already runs BLUE's real engine + 11 hooks + EFSM mirror
+  + live inputs. The day PRIME certifies, **UV v1 = the certified PRIME decision path promoted
+  from shadow to active** (new process/config, journals `dolphin_uv.uv_decisions`, DARK) +
+  a KernelIntent emitter. Zero algorithm code rewritten between certification and first trade.
+- **NO new exec layer.** DITAv2 is already VST-proven (PINK burn-in; VIOLET v4 runner
+  precedent — reuse its KernelIntent→DITAv2 wiring pattern as a NEW instance; never edit the
+  live v4 runner). DITAv2 stays vendored (VENDOR.lock; improvements upstream + vendor_sync).
+  ASEx improvements ride along inside DITAv2's existing integration — the single-writer /
+  no-async-seam guarantees are why it's the kernel; we do not re-plumb them.
+- **NO new reconciler from scratch.** Adopt DITAv2's existing reconcile logic (PINK ownership-
+  filter lineage) under UV's own VST account/clientOrderId prefix (`u-`); C8 shrinks to
+  "verify + adapt + rate-budget test", not "build".
+- **Arming ladder (operator-gated per rung):** DARK (journal-only) → observe-only → VST canary
+  (min size, single slot) → staged size. `ALLOW_MAINNET=0` throughout; mainnet is out of scope.
+- **Decision-vs-execution parity split:** decision layer must bit-match certified-BLUE at
+  Q=scan (differ green in shadow while UV trades); execution quality is judged by
+  reconciler-clean + DITAv2 accounting integrity (fill-price PnL doctrine), since BLUE's
+  "fills" are in-memory bookkeeping and UV's are real VST mechanics.
+- **C11 SL/TP fast clock** (rate-parity stage 2 — protective actions faster than eigenscan,
+  alpha stays at Q=scan): unlocked AFTER first clean canary trades, not before.
+- **Exit = §1 end-state acceptance.** That is the whole spec.
+
+## 9. Deferred (post-first-trade; separate specs — do NOT build in this spec's waves)
+
+- C1 UV-own Zinc plane + generalized Q-dial (UV v1 runs at scan cadence natively — Q=1:1 —
+  so the dial abstraction earns nothing until stage-3 loosening).
+- C2 GraalPy smoke rig; C9 DITAv2 innards probe; TUI beyond the existing panels.
+- Wave-Graal: the GraalPy/GraalVM/Graal-OS migration dev-ops pass (own spec when VST soak
+  is running). Q-loosening experiments (rate-parity stage 3) live there or after.
+
+### 9.1 Sketchlog lane (operator-flagged CRITICAL, 2026-07-02)
+Source: `prod/docs/VIOLET_TODO_CRITICAL_DISTRIBUTION_TRACKING_IN_CONSTRAINED_MEMORY.md`
+(9 signals mapped to BIBLE integration points; sketchlog = DDSketch/HLL/CMS/DriftSketch,
+93 KB constant memory, mergeable monoids, WindowedStreamLog realtime windows, optional C++).
+- **Now (observability, zero parity risk):** sketch dimensions (vel_div percentiles,
+  signal breadth HLL, reversal freq) added to PRIME snapshot + journal as OBSERVE-ONLY
+  columns; T7 replay computes them over full history = instant candidate-feature backtest.
+- **Post-cert (alpha, gated):** signals as decision inputs (esp. #3 rolling-MAE tail
+  detector → adaptive exits = the left-tail killer; #1 widening; #4 breadth; #6 reversal)
+  — UV-divergence features via the certified-Q baseline + diff-guarded rollout. NEVER BLUE.
+- Merge algebra fits PRIME-[n] farms (coordination-free merge); pure-Python path = §10 G1 ok.
+
+### 9.2 Control plane (operator directive 2026-07-02: "NATS/iceoryx2 the hell out of it")
+One control plane over the WHOLE system: fleet lifecycle (start/stop/arm PRIME-[n]/UV/soaks),
+config + Q-dial distribution, heartbeats, gate-ledger events, kill-switch propagation.
+- **Split doctrine:** data plane intra-box = Zinc/iceoryx2 (ADR-1, adopted); CONTROL plane
+  inter-process/inter-box = message bus. Candidate: NATS (operator-named). NOTE: ADR-2
+  reserved Zenoh for inter-box DATA — NATS-for-control vs Zenoh-for-data can coexist;
+  Fable authors the control-plane ADR when wave 2 opens.
+- NOT on the critical path to first testnet trade; REQUIRED before the clone farm.
+- Design-in now (already true): every long-running process publishes a zinc snapshot and
+  takes env-injected config — those are the surfaces the control plane will drive.
+
+## 10. Graal-readiness — build constraints binding NOW despite deferred migration (NFR-G)
+
+Cheap guardrails (mostly "don't do X"), enforced in review — so the later migration is a
+runtime swap, not a rewrite:
+
+- **G1** No CPython-only C-extensions in UV hot paths (pure-Python, or Rust behind stable FFI).
+- **G2** No `__del__`/refcount-timing for correctness — explicit lifecycle (ASEx Drop-reliance
+  leak = the cautionary tale).
+- **G3** No "GIL makes this safe" — cross-thread state only via single-writer/Zinc/ASEx seams.
+- **G4** UV code never imports `hazelcast` directly — HZ quarantined behind the existing
+  reader seams (HZBridge direction).
+- **G5** No hardcoded paths/creds; env-injected config.
+- **G6** Long-running RSS-stable, jemalloc-compatible processes.
+
+## 11. T6 (ACTIVE, HIGH PRIORITY — promoted 2026-07-02): real-Zinc unification
+
+The clone line (banked as branch `salvage/uv-clone-line-1902e4b`) carries the REAL mmap Zinc
+region transport (`ZincShadowChannel`, prefix `uv_shadow`, 18 h live soak) + the Textual
+`tui_v2`. The merged main publishes the same snapshot via atomic file (cross-process, works
+today) + in-memory zinc. T6 = port `ZincShadowChannel` under the reshape snapshot contract as
+the transport, re-home `tui_v2` on it. Small, spec to follow; does NOT gate T1-D relaunch.

prod/docs/UV_TESTNET_ARMING_CHECKLIST.md

@@ -0,0 +1,54 @@
+# UV TESTNET ARMING CHECKLIST — DARK → LIVE on BingX VST
+
+**For the operator.** Execute top to bottom; every box is a stop-if-red. The integrator
+(Fable/successor) signs §1–2; only YOU execute §3.
+
+## §1 — Certification complete (integrator signs)
+- [ ] Gate A verdict PASS in `prod/docs/uv_cert/GATE_AB_RUN_<ts>.md` (Tier 1 + Tier 2
+      green; Tier 3a/3b drift-rates reported and explained). T10 merged.
+- [ ] Gate B verdict PASS (fuzz surface green, deterministic re-run identical).
+- [ ] T2P1 merged: journal writes verified against real DateTime64(3); raw scan surface
+      (`assets[]`, `asset_prices[]`) persisting per scan into dolphin_uv; creds never
+      in URL.
+- [ ] Zinc soak healthy ≥24h on current main: region seq at scan cadence, TUI
+      `source:zinc`, RSS flat (ram block), zero writes outside dolphin_uv (guard log).
+
+## §2 — Seam dry-run reviewed (integrator signs)
+- [ ] T9 merged: 3 injected intents fully mapped + journaled, ZERO venue calls (DARK
+      default proven); `u-` prefix litmus green; mainnet-block mutation test green.
+- [ ] T12 merged: bridge inert-by-default proven; shadow-mode journal shows SUPPRESSED
+      intents recorded during a live soak window — review that excerpt: are these the
+      trades you'd want taken?
+- [ ] End-to-end DARK rehearsal: PRIME live scan → bridge (suppressed) → verify the
+      would-have-been intent in dolphin_uv.exec_journal matches the journaled decision
+      bit-for-bit.
+
+## §3 — Arming (OPERATOR ONLY, in this order)
+1. [ ] Create VST API keys on BingX testnet (never mainnet keys anywhere near this box's
+       UV env). Fund the VST account.
+2. [ ] Place keys per T9's config seam (env/file per merged T9 README — keys never in
+       git, never in the repo tree).
+3. [ ] Start the seam runner (uv_exec instance) STILL DARK — verify it authenticates,
+       reads balances, places NOTHING. Check venue telemetry region populating.
+4. [ ] Write the arming file: `echo "$(date -Iseconds) <your-initials>" >
+       /root/uv-wt/prime-live/UV_PROMOTED.arm` and set `UV_PROMOTED=1` on the PRIME
+       runner env; restart PRIME runner.
+5. [ ] FIRST TRADE WATCH: sit on the TUI for the first bridged intent → order. Verify
+       on the venue: clientOrderId starts `u-`, size/leverage match the journaled
+       intent, position appears in dolphin_uv.exec_journal with venue echo.
+6. [ ] Let it run ONE session. Review: every venue order has a matching BRIDGE row and
+       journaled decision; PnL accounting sane vs venue statement.
+
+## §4 — Kill / rollback (know it BEFORE arming)
+- Instant stop: delete `UV_PROMOTED.arm` (bridge goes inert next scan) — no restart
+  needed. Harder stop: kill the uv_exec runner (positions remain on venue — close via
+  venue UI if needed; it is VST money).
+- Any anomaly (order without journal row, journal row without order, non-`u-` order,
+  any write appearing outside dolphin_uv): kill first, forensics second, report on h5i.
+
+## Standing constraints while live
+- Gate C (live soak) is observation, not a pass/fail blocker — but the T8 stop-watcher
+  ledger and the differ (PRIME still shadows BLUE — parity keeps being measured WHILE
+  UV trades) are the instruments. Divergence trend = stand down and investigate.
+- BLUE remains untouched, unthrottled, unshadowed by any of this. If UV ever competes
+  with BLUE for the same venue account: it must not — separate VST account, always.

prod/docs/uv_subspecs/UV_TASK_T10_CMD_GATE_AB_RUN.md

@@ -0,0 +1,48 @@
+# UV TASK T10 — Gate A/B certification run (the cert that unlocks testnet)
+
+**Assignee:** cmd-PASS1.1 · **Issuer:** Fable · **PRIORITY over T8 resume** (T8 stays
+paused; resume after T10). **Master spec:** §5 Gates A/B. **Base:** main ≥ b70d5ab
+(differ is LIVE — `parity.differ` real module, 33/33 contract tests green).
+
+## Goal
+Produce the first REAL certification verdict: run Gate A (replay parity) and Gate B
+(fuzz faultlines) end-to-end, emit the gate ledger via the T4 reporter, and hand Fable
+a signed-off report. This is the last gate before the operator arms VST keys.
+
+## Work items
+1. **Apply the T7 tier rework** (codex's unapplied instruction, on the bus + here):
+   restructure `replay_cert.py`'s report into labeled coverage tiers —
+   - **Tier 1 (gating):** replay over `dolphin_uv.prime_decisions` (PRIME's own journal,
+     accumulating since the zinc soak began). Today it holds decisions + hook frames
+     (not yet raw scan surface — pi's T2P1 adds that); so Tier 1 v1 = decision-replay
+     consistency: journaled decision == recomputed decision from journaled inputs where
+     inputs suffice; report the input-coverage fraction HONESTLY per row.
+   - **Tier 2 (gating, entries only):** entry-anchored bit-identity — replay each BLUE
+     `dolphin.trade_events` entry through the hooks using its own
+     `market_state_bundle_json` + `execution_quality_json` recorded inputs; diff via
+     `parity.differ.diff_entry_event`. This works over DEEP history NOW.
+   - **Tier 3a (labeled, non-gating):** ~2wk real-scan parquet era (2026-03-04..18,
+     `/mnt/dolphin/vbt_cache_klines`, loader precedent `paper_trade_flow.load_day_scans`)
+     — near-full input replay, report drift-rate.
+   - **Tier 3b (labeled, non-gating):** scalar+obf consistency (eigen_scans scalars +
+     obf_universe prices) — no-phantom-entry checks only.
+2. **Wire the T4 reporter:** gate ledger rows for Gate A (verdict = Tier1 AND Tier2
+   pass) and Gate B; ledger + report to `prod/docs/uv_cert/GATE_AB_RUN_<ts>.md/.json`.
+3. **Gate B run:** execute the full armed property/fuzz surface (pi's Hypothesis suites
+   + differ property tests) with `--hypothesis-seed` pinned, plus a poison-input sweep
+   of the differ/align path (NaN/inf/unicode/dup scans — most exist already; run them
+   as a named gate, record counts). Gate B verdict = all green, deterministic re-run
+   identical.
+4. **Determinism proof:** run the whole cert TWICE; reports must be byte-identical
+   except the run timestamp (seed-pin, injected clock).
+
+## Iron rules
+Read-only against `dolphin.*`; writes ONLY `dolphin_uv.*` + report files. no_write_guard
+active during all replay. Never touch BLUE. If Tier 2 surfaces real DRIFT (not a
+BLUE-code-era artifact per data-derived changepoints): STOP, report to Fable, do NOT
+"fix" hook math to make it pass — parity bugs are findings, not test failures.
+
+## Done
+Branch `uv/t10-gate-ab` off ≥ b70d5ab, fresh clone, push → CI, DONE to Fable with:
+report path + headline (entries checked, matched/drift/phantom per tier, Gate A verdict,
+Gate B verdict, wall-clock). Sample DiffReport for any drift.

prod/docs/uv_subspecs/UV_TASK_T12_PI_PROMOTION_BRIDGE.md

@@ -0,0 +1,40 @@
+# UV TASK T12 — promotion bridge (PRIME decision → KernelIntent → seam)
+
+**Assignee:** pi_nvnemo (AFTER T2P1 — do not context-switch) · **Issuer:** Fable.
+**Master spec:** §8 fast path. **Companion:** T9 exec seam (cmd-PASS1.2, in flight).
+
+## Goal
+The missing link between the certified brain and the trading hand: when PRIME (shadow)
+is PROMOTED, its entry decisions must flow as KernelIntents into T9's seam. Today
+nothing connects them — T9's `intent_source` is an injectable queue. Build the bridge,
+DARK-safe by construction.
+
+## Deliverable: `prod/clean_arch/violet/uv/blue_prime/promotion.py`
+1. **Gate:** module does NOTHING unless `UV_PROMOTED=1` in env AND an operator arming
+   file exists (`/root/uv-wt/prime-live/UV_PROMOTED.arm` — content = ISO ts + operator
+   initials; absence = shadow mode, bridge inert). Two-man rule mirrors T9's mainnet
+   block. Log loudly on every startup which mode we are in.
+2. **Translation:** PRIME's per-scan decision dict (`has_entry`, asset, side, leverage,
+   bar_idx, entry payload — exactly what `build_journal_frame` sees) → `KernelIntent`
+   (dita_v2 contracts; import from the vendored kernel; preflight the exact field set
+   against `contracts.py` — do NOT guess fields). `client_tag` = `u-` prefix source.
+3. **Transport:** publish to the seam's intent queue. v1 = the injectable queue seam
+   T9 exposes (in-process import); leave a clearly-marked TODO seam for zinc
+   intent_region transport (that is wave-2, do not build it now).
+4. **Journal:** every bridged intent (and every SUPPRESSED one while un-promoted) →
+   `dolphin_uv.exec_journal` kind='BRIDGE' rows. The suppressed-intent record is the
+   shadow-mode evidence the operator reviews before arming.
+5. **Runner wiring:** guarded call from `blue_prime/runner.py` after the journal write
+   (never before — journal is the source of truth), wrapped so ANY bridge exception
+   cannot kill the scan loop (fail-soft, log + count).
+
+## Tests (armed and RUN before ship — remember the differ lesson)
+Inert-by-default (no env/file ⇒ zero intents, suppressed rows journaled); arming
+matrix (env-only ⇒ inert, file-only ⇒ inert, both ⇒ active); translation bit-exact
+fixtures vs a fixed decision dict; mutation litmus (drop u- tag, skip suppressed
+journal, swallow exception ⇒ RED); fail-soft (bridge raises ⇒ runner loop continues).
+
+## Done
+Branch `uv/t12-promotion-bridge`, fresh clone, push → CI, DONE to Fable with sha,
+test count, and a shadow-mode journal excerpt showing suppressed intents recorded
+during a live soak window.

prod/docs/uv_subspecs/UV_TASK_T6_CODEX_REAL_ZINC.md

@@ -0,0 +1,32 @@
+# UV TASK T6 — real Zinc region transport (close the file-snapshot regression)
+
+**Assignee:** codex · **Issuer:** Fable · **PRIORITY: HIGH — operator-ordered promotion.**
+**Why:** the merged main's TUI path reads an atomic JSON file in /dev/shm (`ShmChannel`) —
+file semantics, no mapped region/seqlock/notify. That is the exact "mock shm" the shm-reshape
+sprint existed to eliminate. The REAL transport already exists and soaked 18 h on the clone
+line: `ZincShadowChannel` in `salvage/uv-clone-line-1902e4b`'s `uv/shm.py` (mapped
+`/dev/shm/zinc_uv_shadow_state` region, magic header, monotonic seq, `uv_shadow` prefix,
+DITAv2 zinc adapter loading via `ZINC_PYTHON_PATH`).
+
+**Task:** port `ZincShadowChannel` (+ its region encode/decode + `_FileChannel` fallback
+mechanics as needed) from the salvage branch onto current main, UNDER the reshape snapshot
+contract (BluePrimeSnapshot stays the schema — transport changes, contract does not):
+1. New/updated module in `uv/` (e.g. extend `shm.py` or add `zinc_channel.py`): writer =
+   publish the SAME versioned snapshot into the mapped region, atomic + seq-increment +
+   notify per the zinc-shadow spec §5; reader = TUI-side wait/read of latest complete frame.
+2. `blue_prime/runner.py`: publish via real zinc region as PRIMARY; keep the file snapshot as
+   explicit FALLBACK (env `UV_SHM_TRANSPORT=file`; default `zinc`), not the default.
+   [SPEC AMENDED 2026-07-02: knob name aligned to the salvage implementation's
+   `UV_SHM_TRANSPORT` — same semantics as the original `UV_SHM_FALLBACK`, clearer name,
+   zero churn on the already-soaked code.]
+3. `uv/tui.py`: read the region directly as PRIMARY (same render path — snapshot dict in,
+   panels out); file fallback only when region absent, and SAY SO on screen.
+4. Port/adapt the salvage branch's zinc tests + add: torn-frame/seq test, cross-process
+   round-trip test (writer proc + reader proc), region-absent fallback test, mutation litmus.
+5. Relaunch the guarded soak (same UV_BLUE-PRIME_TUI2 discipline, guard active) on the zinc
+   path; verify seq advances at scan cadence and TUI reads the REGION (show source in meta).
+**Non-goals:** no tui_v2 port (separate, later), no DITAv2 region reuse (own prefix,
+retire-able — zinc-shadow spec §3.3), no schema changes.
+**Branch:** `uv/t6-real-zinc` · fresh clone `/root/uv-wt/t6-zinc` off /root/violet.git ·
+push→CI · DONE to Fable with: branch+sha, tests, soak PIDs + region seq evidence, and
+`grep`-proof the TUI's primary path never opens the JSON file when the region exists.

prod/docs/uv_subspecs/UV_TASK_T8_CMD_STOP_WATCHER.md

@@ -0,0 +1,49 @@
+# UV TASK T8 — shadow stop-watcher (C11's sensor, live overshoot ledger)
+
+**Assignee:** cmd (Command Code) · **Issuer:** Fable · **Master spec:** §4 C11, §5 Gate C.
+**Empirical basis:** `prod/docs/BLUE_STOPLOSS_OVERSHOOT_AND_UV_COUNTERFACTUAL_20260702.md`
+(codex, verified by Fable): last-1000 BLUE trades → 49 stops, 30 overshot the −1.2%
+contract, ≈$4,432 excess. FET `e81e595d`: stop breached 16:08:15.9, eigenscan not until
+16:08:19.6 (measured gaps 11–12 s), overshoot lived entirely inside the scan gap.
+
+## Goal
+A standalone read-only daemon that watches BLUE's OPEN positions on a ~1 s clock and
+journals every stop-contract breach the moment it happens — so every future overshoot is
+measured live instead of forensically. This is the shadow twin of the future C11 fast SL
+clock: when UV trades with fast stops, this ledger IS the A/B evidence.
+
+## Deliverable: `prod/clean_arch/violet/uv/stop_watcher.py` (+ DDL + tests)
+1. **Position feed:** poll `dolphin.trade_events` (read-only, dedup by trade_id, argMax ts)
+   every ~10 s for open positions (entry event without terminal exit). Carry entry_price,
+   side, quantity, our_leverage.
+2. **Price feed:** poll `dolphin.obf_universe` (read-only) best_bid/best_ask per open asset
+   every ~1 s. MEASURE ingest lag (row ts vs now) and journal it; if p95 lag > 2 s, log a
+   LIMITATION line — do NOT silently trust stale prices. (Fallback to BingX public WS
+   bookTicker is allowed — public data, no keys, no BLUE interaction — but is a stretch
+   goal, not v1.)
+3. **Breach detection:** SHORT: ask ≥ entry×1.012; LONG: bid ≤ entry×0.988. On first
+   breach per trade, write one event; keep sampling and write escalation rows at each
+   +0.1 % beyond the stop (so the overshoot PATH is recorded, not just the edge).
+4. **Journal:** `dolphin_uv.stop_watch_events` (NEW table, dolphin_uv namespace ONLY):
+   ts, trade_id, asset, side, entry_price, breach_price, adverse_pct, spread_bps,
+   depth_1pct_usd, obf_lag_ms, kind ('BREACH'|'ESCALATION'|'RESOLVED'). RESOLVED row when
+   the trade's exit appears in trade_events, carrying exit adverse_pct + excess vs 1.2 %.
+5. **Daily rollup view:** overshoot count, total excess $, worst trade — the $4.4K audit,
+   automated forever.
+
+## Iron rules
+- ZERO writes outside `dolphin_uv.*`. Never touch BLUE code, HZ contents, or dolphin.*
+  tables. CH creds read-only usage; INSERT only into dolphin_uv.stop_watch_events.
+- Runs under /home/dolphin/siloqy_env, own log file, no hardcoded worktree paths
+  (derive paths from __file__ / env).
+
+## Tests (doctrine: mutation litmus mandatory)
+Synthetic price/position fixtures → exact expected breach/escalation/resolved sequence;
+mutation litmus (flip breach comparison, drop escalation step → tests go RED);
+lag-measurement honesty test; no-write guard test (any non-dolphin_uv INSERT raises);
+determinism (same fixture ⇒ same journal twice).
+
+## Done
+Branch `uv/t8-stop-watcher` off current main, fresh clone, push → CI, DONE to Fable with:
+branch+sha, test count, a live 30-min run's journal excerpt (real breaches or honest
+"no breaches in window"), measured obf_universe lag stats.

prod/docs/uv_subspecs/UV_TASK_T9_CMD_DITAV2_EXEC_SEAM.md

@@ -0,0 +1,48 @@
+# UV TASK T9 — C10 DITAv2 exec seam (KernelIntent → DITAv2 → BingX VST, DARK)
+
+**Assignee:** cmd-PASS1.2 · **Issuer:** Fable · **Master spec:** §4 C10, §8 fast path.
+**Doctrinal kernel:** post-sync vendored dita_v2 per `UV_DITAV2_SOA_VERDICT_20260703.md`
+(includes VenueTelemetrySnapshot + zinc venue plane + asex_account.py). Do NOT start
+until Fable confirms the vendor sync landed on main (watch the bus).
+
+## Goal
+The physical trading path: a UV-owned execution runner that consumes KernelIntents and
+drives DITAv2 against BingX **VST testnet** — built now, DARK by default, so that when
+PRIME is promoted (post Gates A+B) the only remaining step is arming it with keys.
+Pattern precedent: `prod/clean_arch/violet/v4_execution_runner.py` (read for shape; this
+is a NEW instance and a NEW module — never reuse a BLUE/PINK/VIOLET runner instance).
+
+## Deliverable: `prod/clean_arch/violet/uv/exec/` (new package)
+1. **`intent_source.py`:** KernelIntent inlet. v1 = injectable queue + a file/CLI
+   injector for dry-run intents (PRIME promotion wiring is a LATER task — leave a
+   clearly-marked seam, not a stub that pretends).
+2. **`seam.py`:** intent → DITAv2 order mapping. EVERY clientOrderId prefixed `u-`
+   (non-negotiable — this is how UV's orders are distinguishable on the venue forever).
+   Sizing/leverage passthrough from intent; no local overrides.
+3. **`runner.py`:** launcher wiring per dita_v2 `launcher.py` — NEW instance name
+   (`uv_exec`), zinc venue plane ENABLED (venue_region telemetry is the seam's flight
+   recorder), jemalloc-friendly long-run posture, no hardcoded paths.
+4. **DARK doctrine:** with no keys configured → observe-only: log + journal every intent
+   and the order it WOULD place (full params), place nothing. `ALLOW_MAINNET=0` is a
+   hard block: mainnet refuses even if env says otherwise unless a separate operator
+   arming file exists (two-man rule). VST base URL only.
+5. **Journal:** every intent, mapping, would-place/placed, venue telemetry snapshot →
+   `dolphin_uv.exec_journal` (dolphin_uv namespace ONLY).
+
+## Iron rules
+- VST ONLY. DARK until operator arms. `u-` prefix on every clientOrderId.
+- Zero writes outside dolphin_uv.*; zero BLUE touches; vendored dita_v2 is read-only
+  (any kernel change goes upstream + vendor_sync, never in-place).
+- Graal-ready NFR-G applies (no CPython-only exotica in the seam layer).
+
+## Tests (mutation litmus mandatory)
+Intent→order mapping bit-exact fixtures; `u-` prefix litmus (strip the prefix in code →
+test RED); DARK default test (no keys ⇒ zero venue calls — assert at the venue adapter
+seam, not by mocking the seam itself); mainnet-block mutation test (force
+ALLOW_MAINNET=1 without arming file ⇒ still refuses); venue-plane telemetry presence
+test; determinism.
+
+## Done
+Branch `uv/t9-exec-seam` off post-sync main, fresh clone (NOT a worktree of the bare),
+push → CI, DONE to Fable with: branch+sha, test count, and a dry-run journal excerpt
+showing 3 injected intents fully mapped + journaled + zero venue calls.

test_pi_wake_agent.py

@@ -0,0 +1,304 @@
+#!/usr/bin/env python3
+"""
+Comprehensive test suite for pi_wake_agent.py
+Tests all modes, edge cases, and the new succession feature.
+"""
+
+import pytest
+import subprocess
+import time
+import tempfile
+import os
+import sys
+from pathlib import Path
+from unittest.mock import patch, MagicMock, call
+
+# Add the script directory to path
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+from pi_wake_agent import (
+    parse_interval,
+    interval_to_cron,
+    interval_to_human,
+    cron_comment,
+    parse_sessions,
+    SCRIPT_PATH,
+    LOG_FILE,
+    LOG_MAX_SIZE,
+    LOG_MAX_FILES,
+    CRON_COMMENT_PREFIX,
+    AGENT_NICK,
+    H5I_AGENT,
+    H5I_BUS_ROOT,
+    DEFAULT_INTERVAL,
+)
+
+# ─── Test parse_interval ─────────────────────────────────────────────────
+
+class TestParseInterval:
+    def test_hours(self):
+        assert parse_interval("1h") == 3600
+        assert parse_interval("2h") == 7200
+        assert parse_interval("24h") == 86400
+    
+    def test_minutes(self):
+        assert parse_interval("30m") == 1800
+        assert parse_interval("1m") == 60
+        assert parse_interval("90m") == 5400
+    
+    def test_seconds(self):
+        assert parse_interval("30s") == 30
+        assert parse_interval("1s") == 1
+        assert parse_interval("10s") == 10
+    
+    def test_invalid(self):
+        with pytest.raises(ValueError):
+            parse_interval("invalid")
+        with pytest.raises(ValueError):
+            parse_interval("1x")
+        with pytest.raises(ValueError):
+            parse_interval("")
+
+# ─── Test interval_to_cron ───────────────────────────────────────────────
+
+class TestIntervalToCron:
+    def test_hours(self):
+        assert interval_to_cron("1h") == "0 */1 * * *"
+        assert interval_to_cron("2h") == "0 */2 * * *"
+        assert interval_to_cron("6h") == "0 */6 * * *"
+    
+    def test_minutes(self):
+        assert interval_to_cron("1m") == "*/1 * * * *"
+        assert interval_to_cron("30m") == "*/30 * * * *"
+        assert interval_to_cron("45m") == "*/45 * * * *"
+    
+    def test_invalid_minutes(self):
+        with pytest.raises(ValueError):
+            interval_to_cron("60m")
+        with pytest.raises(ValueError):
+            interval_to_cron("90m")
+    
+    def test_invalid_seconds(self):
+        with pytest.raises(ValueError):
+            interval_to_cron("30s")
+    
+    def test_invalid_format(self):
+        with pytest.raises(ValueError):
+            interval_to_cron("invalid")
+
+# ─── Test interval_to_human ──────────────────────────────────────────────
+
+class TestIntervalToHuman:
+    def test_hours(self):
+        assert interval_to_human("1h") == "1 hour(s)"
+        assert interval_to_human("2h") == "2 hour(s)"
+    
+    def test_minutes(self):
+        assert interval_to_human("30m") == "30 minute(s)"
+        assert interval_to_human("1m") == "1 minute(s)"
+    
+    def test_seconds(self):
+        assert interval_to_human("30s") == "30 second(s)"
+        assert interval_to_human("1s") == "1 second(s)"
+    
+    def test_invalid(self):
+        assert interval_to_human("invalid") == "invalid"
+
+# ─── Test cron_comment ───────────────────────────────────────────────────
+
+class TestCronComment:
+    def test_single_session(self):
+        sessions = ["cc_UV_dev0_Fb"]
+        result = cron_comment(sessions, "1h")
+        assert result == "pi_wake_agent:cc_UV_dev0_Fb:1h"
+    
+    def test_multiple_sessions(self):
+        sessions = ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
+        result = cron_comment(sessions, "30m")
+        assert result == "pi_wake_agent:cc_UV_dev0_Fb,cc_UV_dev1_48:30m"
+    
+    def test_empty_sessions(self):
+        result = cron_comment([], "1h")
+        assert result == "pi_wake_agent::1h"
+
+# ─── Test parse_sessions ─────────────────────────────────────────────────
+
+class TestParseSessions:
+    def test_comma_separated(self):
+        result = parse_sessions("cc_UV_dev0_Fb,cc_UV_dev1_48")
+        assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
+    
+    def test_single_session(self):
+        result = parse_sessions("cc_UV_dev0_Fb")
+        assert result == ["cc_UV_dev0_Fb"]
+    
+    def test_with_spaces(self):
+        result = parse_sessions("cc_UV_dev0_Fb, cc_UV_dev1_48")
+        assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
+    
+    def test_empty(self):
+        assert parse_sessions("") == []
+        assert parse_sessions(None) == []
+    
+    def test_list_input(self):
+        result = parse_sessions(["a", "b", "c"])
+        assert result == ["a", "b", "c"]
+    
+    def test_filters_empty(self):
+        result = parse_sessions("a,,b")
+        assert result == ["a", "b"]
+
+# ─── Test main functions via subprocess ──────────────────────────────────
+
+SCRIPT = "/mnt/dolphinng5_predict/pi_wake_agent.py"
+
+def run_cmd(args, timeout=30):
+    """Run the script and return (returncode, stdout, stderr)"""
+    result = subprocess.run(
+        [sys.executable, SCRIPT] + args,
+        capture_output=True,
+        text=True,
+        timeout=timeout
+    )
+    return result.returncode, result.stdout, result.stderr
+
+# ─── Integration Tests ──────────────────────────────────────────────────
+
+class TestHelp:
+    def test_help(self):
+        rc, out, err = run_cmd(["--help"])
+        assert rc == 0
+        assert "pi_wake_agent.py" in out
+        assert "--install" in out
+        assert "--once" in out
+        assert "--daemon" in out
+        assert "--succession" in out
+        assert "--count" in out
+
+class TestList:
+    def test_list_no_cron(self):
+        # Clear any existing cron entries first
+        subprocess.run(["crontab", "-l"], capture_output=True)
+        subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
+        
+        rc, out, err = run_cmd(["--list"])
+        assert rc == 0
+        assert "pi_wake_agent cron entries" in out
+
+class TestInstallRemove:
+    def setup_method(self):
+        # Clear cron before each test
+        subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
+    
+    def teardown_method(self):
+        subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
+    
+    def test_install_and_list(self):
+        rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "test_session", "--msg", "test"])
+        assert rc == 0
+        
+        rc, out, err = run_cmd(["--list"])
+        assert rc == 0
+        assert "test_session" in out
+        assert "1h" in out
+        
+        # Cleanup
+        run_cmd(["--remove", "--session", "test_session", "--interval", "1h"])
+    
+    def test_install_multi_session(self):
+        rc, out, err = run_cmd(["--install", "--interval", "30m", "--sessions", "s1,s2", "--msg", "multi"])
+        assert rc == 0
+        
+        rc, out, err = run_cmd(["--list"])
+        assert rc == 0
+        assert "s1,s2" in out
+        
+        run_cmd(["--remove", "--sessions", "s1,s2", "--interval", "30m"])
+
+class TestValidate:
+    def test_validate_existing(self):
+        # The test session might not exist, so we test the command runs
+        rc, out, err = run_cmd(["--validate", "--session", "pi_test"])
+        assert rc == 0  # Should not crash
+
+class TestOnce:
+    def test_once_short(self):
+        # Use a very short interval
+        rc, out, err = run_cmd(["--once", "--interval", "1s", "--session", "test_session", "--msg", "quick test"], timeout=10)
+        assert rc == 0
+        # Check log file for the message
+        time.sleep(2)
+        log_content = Path("/tmp/pi_wake_agent.log").read_text()
+        assert "One-shot timer set" in log_content
+
+class TestSuccession:
+    def test_succession_short(self):
+        # Test succession mode with very short intervals
+        rc, out, err = run_cmd([
+            "--succession", "--count", "2", "--interval", "1s",
+            "--session", "test_session", "--msg", "succession test"
+        ], timeout=60)
+        assert rc == 0
+        # Check logs
+        time.sleep(3)
+        log_content = Path("/tmp/pi_wake_agent.log").read_text()
+        assert "SUCCESSION START" in log_content
+        assert "SUCCESSION COMPLETE" in log_content
+    
+    def test_succession_invalid_count(self):
+        rc, out, err = run_cmd(["--succession", "--count", "0", "--interval", "1s", "--session", "test"])
+        # Should fail with invalid count
+        assert rc != 0 or "error" in err.lower()
+
+class TestStatus:
+    def test_status(self):
+        rc, out, err = run_cmd(["--status"])
+        assert rc == 0
+        assert "pi_wake_agent Status" in out
+        assert "Script:" in out
+
+class TestRun:
+    def test_run_mode(self):
+        # This is the internal mode called by cron
+        rc, out, err = run_cmd(["--run", "--session", "test_session", "--msg", "test"])
+        assert rc == 0  # Should succeed even if session doesn't exist
+
+class TestEdgeCases:
+    def test_invalid_interval(self):
+        rc, out, err = run_cmd(["--install", "--interval", "invalid", "--session", "test"])
+        assert rc != 0
+    
+    def test_missing_session(self):
+        rc, out, err = run_cmd(["--install", "--interval", "1h"])
+        assert rc != 0
+        assert "session" in err.lower() or "required" in err.lower()
+    
+    def test_invalid_count(self):
+        rc, out, err = run_cmd(["--succession", "--count", "-1", "--interval", "1s", "--session", "test"])
+        assert rc != 0
+
+class TestSessionParsing:
+    def test_multiple_session_flags(self):
+        rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "s1", "--session", "s2", "--msg", "test"])
+        assert rc == 0
+        run_cmd(["--remove", "--session", "s1", "--interval", "1h"])
+        run_cmd(["--remove", "--session", "s2", "--interval", "1h"])
+
+class TestRemove:
+    def test_remove_nonexistent(self):
+        # Should not crash
+        rc, out, err = run_cmd(["--remove", "--session", "nonexistent", "--interval", "1h"])
+        assert rc == 0
+
+# ─── Test Logging ─────────────────────────────────────────────────────────
+
+class TestLogging:
+    def test_log_file_created(self):
+        run_cmd(["--install", "--interval", "1h", "--session", "log_test", "--msg", "test"])
+        assert LOG_FILE.exists()
+        run_cmd(["--remove", "--session", "log_test", "--interval", "1h"])
+
+# ─── Run all tests ────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v", "--tb=short"])