DOCS: expand recent Violet touched files inventory

#1 (validation) — VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md documents the
ground truth from BLUE's own code (esf_alpha_orchestrator / adaptive_circuit_breaker
/ nautilus_event_trader): of the 8 sizing inputs, ONLY `posture` is a flat HZ key.
`esof_score` is in HZ but as a payload to parse. The other five — boost, beta,
mc_scale, ob_median_imbalance, ob_agreement_pct, dc_status — are BLUE-organ outputs
(ACB over DOLPHIN_FEATURES.exf_latest, MC flag derivation, OBFeatureEngine, the
per-asset signal generator) and are NOT present as scalars in any HZ map. This
inverts live_factors.py's flat-snapshot premise; its speculative alternate paths
(acb_boost / s_acb_boost / ("acb","boost") / …) match nothing real. The full live
sourcing of the five is a multi-organ sprint (V3.4c), not a HZ scrape.

#2 (sourcing adapter) — live_factor_source.py sources what BLUE actually publishes,
read-only and pure (callers pass already-fetched HZ blobs; no client, no I/O):
  - posture     ← engine_snapshot['posture'] (DOLPHIN_STATE_BLUE), default APEX
  - esof_score  ← DOLPHIN_FEATURES['esof_latest'] via BLUE's OWN parse_esof_payload
                  + esof_score_from_payload (wrap, don't reimplement; staleness gate
                  honored when max_age_s supplied)
  - boost/beta/mc_scale/ob_*/dc_status ← BLUE's neutral sentinels (1.0/0.0/1.0/None/
                  None/"NONE") until V3.4c — explicit, never silently faked.
Flows through the validated extract_live_sizing_factors normalizer. ORGAN_DERIVED_
FACTORS names the six deferred to V3.4c so the journal can mark them NEUTRAL.

9 new tests green (posture default/upper, esof dict+raw-JSON+staleness, neutral
integration, all-neutral-when-empty). violet-only; no shared-file edits; no soak.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

prod/clean_arch/violet/live_blue_source.py

@@ -0,0 +1,361 @@
+"""VIOLET V3.4c: read BLUE-published live organs from Hazelcast.
+
+This is VIOLET-only. BLUE is untouched.
+
+The adapter reads the published BLUE surfaces that already exist in HZ and
+translates them into ``SizingFactors`` for the shadow path:
+  - ``posture`` from ``DOLPHIN_STATE_BLUE.latest_nautilus`` / ``engine_snapshot``
+  - ``esof_score`` from ``DOLPHIN_FEATURES.esof_latest`` or ``esof_advisor_latest``
+  - ``acb_boost`` / ``acb_beta`` from ``DOLPHIN_FEATURES.acb_boost``
+  - ``mc_scale`` from ``DOLPHIN_FEATURES.mc_forewarner_latest``
+  - OB market consensus from the live ``asset_*_ob`` maps via BLUE's own
+    ``OBFeatureEngine``
+
+The remaining DC signal is left neutral here for now. It needs the same live
+signal-history path BLUE uses and should be added as a separate mirror step.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from collections import deque
+from collections.abc import Mapping
+from dataclasses import dataclass
+from dataclasses import field
+from pathlib import Path
+from typing import Any, Deque, Iterable, Optional
+
+import hazelcast
+import numpy as np
+
+_PROJECT_ROOT = Path(__file__).resolve().parents[3]
+for _p in (str(_PROJECT_ROOT), str(_PROJECT_ROOT / "nautilus_dolphin")):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from nautilus_dolphin.nautilus.ob_features import OBFeatureEngine
+from nautilus_dolphin.nautilus.ob_provider import OBSnapshot, OBProvider
+from nautilus_dolphin.nautilus.alpha_signal_generator import AlphaSignalGenerator
+
+from .alpha_wrappers import VioletAssetSelector
+from .decision_engine import SizingFactors
+from .live_factor_source import esof_score_from_features, posture_from_engine_snapshot
+from .live_factors import extract_live_sizing_factors
+
+
+def _jsonish(value: Any) -> Any:
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except Exception:
+            return value
+    return value
+
+
+def _coerce_float(value: Any, default: Optional[float] = None) -> Optional[float]:
+    try:
+        if value is None:
+            return default
+        out = float(value)
+        if not np.isfinite(out):
+            return default
+        return out
+    except (TypeError, ValueError):
+        return default
+
+
+def _read_hz_map(client: hazelcast.HazelcastClient, map_name: str, key: str) -> Any:
+    try:
+        return client.get_map(map_name).blocking().get(key)
+    except Exception:
+        return None
+
+
+def _map_status_to_mc_scale(payload: Any) -> float:
+    data = _jsonish(payload)
+    if isinstance(data, Mapping):
+        status = str(data.get("status", "")).upper()
+    else:
+        status = str(data).upper()
+    return 0.5 if status == "ORANGE" else 1.0
+
+
+def _extract_acb(payload: Any) -> tuple[float, float]:
+    data = _jsonish(payload)
+    if isinstance(data, Mapping):
+        boost = _coerce_float(data.get("boost"), 1.0) or 1.0
+        beta = _coerce_float(data.get("beta"), 0.0) or 0.0
+        return max(0.0, boost), max(0.0, beta)
+    return 1.0, 0.0
+
+
+def _scan_view(payload: Any) -> Mapping[str, Any]:
+    """Normalize NG7 nested or NG8 flat scan payloads into one mapping."""
+    data = _jsonish(payload)
+    if not isinstance(data, Mapping):
+        return {}
+    view: dict[str, Any] = dict(data)
+    result = data.get("result")
+    if isinstance(result, Mapping):
+        view.update(result)
+    return view
+
+
+def _scan_assets(scan: Mapping[str, Any]) -> list[str]:
+    assets = scan.get("assets")
+    if isinstance(assets, list) and assets:
+        return [str(a).upper() for a in assets if a is not None]
+    target = scan.get("target_asset") or scan.get("asset")
+    return [str(target).upper()] if target else []
+
+
+def _scan_prices(scan: Mapping[str, Any]) -> list[float]:
+    prices = scan.get("asset_prices") or scan.get("prices")
+    if not isinstance(prices, list):
+        return []
+    out: list[float] = []
+    for value in prices:
+        px = _coerce_float(value, None)
+        if px is None or px <= 0:
+            continue
+        out.append(px)
+    return out
+
+
+@dataclass
+class LiveBlueScanHistory:
+    """Stateful scan history mirror for DC confirmation.
+
+    BLUE computes `dc_status` from the active asset price history. This helper keeps
+    a read-only replay of the published scan stream so VIOLET can reproduce that
+    state without touching BLUE or inventing a new schema.
+    """
+
+    maxlen: int = 32
+    trade_direction: int = -1
+    _prices: dict[str, Deque[float]] = field(default_factory=dict)
+    last_scan_number: Optional[int] = None
+
+    def ingest_scan(self, scan_payload: Any) -> Mapping[str, Any]:
+        scan = _scan_view(scan_payload)
+        assets = _scan_assets(scan)
+        prices = _scan_prices(scan)
+        for asset, price in zip(assets, prices):
+            if price <= 0:
+                continue
+            hist = self._prices.setdefault(asset, deque(maxlen=self.maxlen))
+            hist.append(float(price))
+        sn = _coerce_float(scan.get("scan_number"), None)
+        if sn is not None:
+            self.last_scan_number = int(sn)
+        return scan
+
+    def price_history(self, asset: str) -> list[float]:
+        return list(self._prices.get(str(asset).upper(), ()))
+
+    def market_data(self, lookback: int) -> dict[str, list[float]]:
+        need = max(1, int(lookback) + 1)
+        return {asset: list(hist) for asset, hist in self._prices.items() if len(hist) >= need}
+
+    def dc_status(self, scan_payload: Any, *, already_ingested: bool = False) -> str:
+        scan = _scan_view(scan_payload) if already_ingested else self.ingest_scan(scan_payload)
+        asset = _scan_assets(scan)
+        if not asset:
+            return "NONE"
+        vel_div = _coerce_float(scan.get("vel_div"), None)
+        if vel_div is None:
+            return "NONE"
+        history = self.price_history(asset[0])
+        if not history:
+            return "NONE"
+        signal_gen = AlphaSignalGenerator()
+        sig = signal_gen.generate(
+            vel_div=vel_div,
+            vel_div_history=None,
+            asset_price_history=history,
+            trade_direction=self.trade_direction,
+            asset=asset[0],
+            current_timestamp=_coerce_float(scan.get("timestamp"), 0.0) or 0.0,
+        )
+        return sig.dc_status
+
+
+class HazelcastOBProvider(OBProvider):
+    """Read the current BLUE OB shards directly from Hazelcast."""
+
+    def __init__(self, client: hazelcast.HazelcastClient):
+        self.client = client
+
+    def _asset_keys(self) -> list[str]:
+        try:
+            keys = self.client.get_map("DOLPHIN_FEATURES").blocking().key_set()
+        except Exception:
+            return []
+        assets = []
+        for key in keys:
+            if not isinstance(key, str) or not key.startswith("asset_") or not key.endswith("_ob"):
+                continue
+            asset = key[len("asset_"):-len("_ob")]
+            if asset and asset not in assets:
+                assets.append(asset)
+        return sorted(assets)
+
+    def _read_snapshot(self, asset: str) -> Optional[OBSnapshot]:
+        raw = _read_hz_map(self.client, "DOLPHIN_FEATURES", f"asset_{asset}_ob")
+        data = _jsonish(raw)
+        if not isinstance(data, Mapping):
+            return None
+        bid_notional = np.array(
+            [_coerce_float(v, 0.0) or 0.0 for v in data.get("bid_notional", [0, 0, 0, 0, 0])][:5],
+            dtype=np.float64,
+        )
+        ask_notional = np.array(
+            [_coerce_float(v, 0.0) or 0.0 for v in data.get("ask_notional", [0, 0, 0, 0, 0])][:5],
+            dtype=np.float64,
+        )
+        bid_depth = np.array(
+            [_coerce_float(v, 0.0) or 0.0 for v in data.get("bid_depth", [0, 0, 0, 0, 0])][:5],
+            dtype=np.float64,
+        )
+        ask_depth = np.array(
+            [_coerce_float(v, 0.0) or 0.0 for v in data.get("ask_depth", [0, 0, 0, 0, 0])][:5],
+            dtype=np.float64,
+        )
+        ts = _coerce_float(data.get("timestamp"), 0.0) or 0.0
+        if (
+            bid_notional.shape != (5,) or ask_notional.shape != (5,)
+            or bid_depth.shape != (5,) or ask_depth.shape != (5,)
+        ):
+            return None
+        if np.any(bid_notional < 0) or np.any(ask_notional < 0):
+            return None
+        if np.any(bid_depth < 0) or np.any(ask_depth < 0):
+            return None
+        return OBSnapshot(
+            timestamp=ts,
+            asset=asset,
+            bid_notional=bid_notional,
+            ask_notional=ask_notional,
+            bid_depth=bid_depth,
+            ask_depth=ask_depth,
+        )
+
+    def get_snapshot(self, asset: str, timestamp: float) -> Optional[OBSnapshot]:
+        return self._read_snapshot(asset)
+
+    def get_assets(self) -> list[str]:
+        return self._asset_keys()
+
+    def get_all_timestamps(self, asset: str) -> np.ndarray:
+        snap = self._read_snapshot(asset)
+        if snap is None:
+            return np.array([], dtype=np.float64)
+        return np.array([snap.timestamp], dtype=np.float64)
+
+    def get_snapshot_count(self, asset: str) -> int:
+        return 1 if self._read_snapshot(asset) is not None else 0
+
+
+@dataclass(frozen=True)
+class LiveBlueSourceResult:
+    factors: SizingFactors
+    acb_boost: float
+    acb_beta: float
+    mc_scale: float
+    posture: str
+    dc_status: str
+    selected_asset: str
+
+
+def source_live_blue_sizing_factors(
+    client: hazelcast.HazelcastClient,
+    *,
+    assets: Optional[Iterable[str]] = None,
+    scan_history: Optional[LiveBlueScanHistory] = None,
+    selector: Optional[VioletAssetSelector] = None,
+) -> LiveBlueSourceResult:
+    """Read the BLUE-published live surfaces and return a typed factor plane."""
+    scan_history = scan_history or LiveBlueScanHistory()
+    selector = selector or VioletAssetSelector()
+
+    engine_snapshot_raw = _read_hz_map(client, "DOLPHIN_STATE_BLUE", "latest_nautilus")
+    if engine_snapshot_raw is None:
+        engine_snapshot_raw = _read_hz_map(client, "DOLPHIN_STATE_BLUE", "engine_snapshot")
+    engine_snapshot = _jsonish(engine_snapshot_raw)
+    if isinstance(engine_snapshot, str):
+        try:
+            engine_snapshot = json.loads(engine_snapshot)
+        except Exception:
+            engine_snapshot = {}
+
+    posture = posture_from_engine_snapshot(engine_snapshot if isinstance(engine_snapshot, Mapping) else None)
+
+    esof_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "esof_latest")
+    if esof_raw is None:
+        esof_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "esof_advisor_latest")
+    esof_score = esof_score_from_features(esof_raw)
+
+    acb_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "acb_boost")
+    if acb_raw is None:
+        acb_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "acb_boost_short")
+    acb_boost, acb_beta = _extract_acb(acb_raw)
+
+    mc_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "mc_forewarner_latest")
+    mc_scale = _map_status_to_mc_scale(mc_raw)
+
+    scan_raw = _read_hz_map(client, "DOLPHIN_FEATURES", "latest_eigen_scan")
+    scan = scan_history.ingest_scan(scan_raw)
+    scan_assets = _scan_assets(scan)
+    trade_direction = int(
+        _coerce_float(
+            engine_snapshot.get("trade_direction_runtime") if isinstance(engine_snapshot, Mapping) else None,
+            None,
+        )
+        or _coerce_float(
+            engine_snapshot.get("trade_direction_base") if isinstance(engine_snapshot, Mapping) else None,
+            None,
+        )
+        or -1
+    )
+    candidate_market = scan_history.market_data(selector.lookback)
+    pick = selector.pick(candidate_market, regime_direction=trade_direction)
+    selected_asset = pick.asset if pick is not None else (scan_assets[0] if scan_assets else "")
+    dc_status = scan_history.dc_status(scan, already_ingested=True)
+
+    ob_provider = HazelcastOBProvider(client)
+    ob_engine = OBFeatureEngine(ob_provider)
+    ob_assets = list(assets) if assets is not None else (scan_assets or ob_provider.get_assets())
+    if ob_assets:
+        try:
+            ob_engine.step_live(ob_assets, bar_idx=0)
+            market = ob_engine.get_market(0, ob_assets)
+            ob_median_imbalance = float(market.median_imbalance)
+            ob_agreement_pct = float(market.agreement_pct)
+        except Exception:
+            ob_median_imbalance = None
+            ob_agreement_pct = None
+    else:
+        ob_median_imbalance = None
+        ob_agreement_pct = None
+
+    hz_snapshot = {
+        "boost": acb_boost,
+        "beta": acb_beta,
+        "mc_scale": mc_scale,
+        "esof_score": esof_score,
+        "ob_median_imbalance": ob_median_imbalance,
+        "ob_agreement_pct": ob_agreement_pct,
+        "dc_status": dc_status,
+        "posture": posture,
+    }
+    factors = extract_live_sizing_factors(hz_snapshot=hz_snapshot)
+    return LiveBlueSourceResult(
+        factors=factors,
+        acb_boost=acb_boost,
+        acb_beta=acb_beta,
+        mc_scale=mc_scale,
+        posture=posture,
+        dc_status=dc_status,
+        selected_asset=selected_asset,
+    )

prod/clean_arch/violet/live_factor_source.py

@@ -0,0 +1,123 @@
+"""VIOLET V3.4b: source live ``SizingFactors`` from BLUE's published HZ planes.
+
+The field-path validation (``prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md``)
+established that of the eight sizing inputs, only ``posture`` and ``esof_score`` are
+present in maps live BLUE publishes to Hazelcast:
+
+  - ``posture``     ← ``DOLPHIN_STATE_BLUE`` ``engine_snapshot['posture']``
+  - ``esof_score``  ← ``DOLPHIN_FEATURES['esof_latest'|'esof_advisor_latest']``,
+                      parsed by BLUE's OWN ``parse_esof_payload`` /
+                      ``esof_score_from_payload`` (wrap, don't reimplement).
+
+The remaining five (``boost``, ``beta``, ``mc_scale``, ``ob_median_imbalance``,
+``ob_agreement_pct``, ``dc_status``) are BLUE-organ outputs — the ACB over
+``DOLPHIN_FEATURES['exf_latest']``, the MC flag→scale derivation, ``OBFeatureEngine``,
+and the per-asset signal generator — and are NOT present as scalars in any HZ map.
+Sourcing them live is the V3.4c organ-wiring sprint.
+
+Until then this adapter sources the two HZ-available factors faithfully and supplies
+BLUE's OWN neutral sentinels for the organ-derived five (``boost=1.0``, ``beta=0.0``,
+``mc_scale=1.0``, ``ob_*=None``, ``dc_status="NONE"``). The result flows through the
+validated ``extract_live_sizing_factors`` normalizer, so the V3.4 shadow breakdown
+records posture+esof LIVE and the rest NEUTRAL — explicit, never silently faked.
+
+Pure boundary: callers pass already-fetched HZ blobs (the DARK service reads the maps
+and hands the dicts in). No Hazelcast client here, no I/O, no launcher coupling —
+mirrors ``live_factors.py``'s philosophy and keeps the adapter fully unit-testable.
+"""
+
+from __future__ import annotations
+
+import sys
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any, Optional
+
+from .decision_engine import SizingFactors
+from .domain import typed
+from .live_factors import extract_live_sizing_factors
+
+_PROJECT_ROOT = Path(__file__).resolve().parents[3]
+
+# The organ-derived factors this adapter cannot source from HZ yet (V3.4c). Listed so
+# the journal/diagnostics can mark them NEUTRAL rather than mistaking them for live.
+ORGAN_DERIVED_FACTORS = (
+    "boost", "beta", "mc_scale",
+    "ob_median_imbalance", "ob_agreement_pct", "dc_status",
+)
+
+
+def _import_esof_gate() -> Any:
+    """Import BLUE's ``esof_size_gate`` (same root-injection as ``sizing.py``)."""
+    try:
+        from nautilus_dolphin.nautilus import esof_size_gate  # type: ignore
+    except ImportError:
+        for p in (str(_PROJECT_ROOT / "nautilus_dolphin"), str(_PROJECT_ROOT)):
+            if p not in sys.path:
+                sys.path.insert(0, p)
+        sys.modules.pop("nautilus_dolphin", None)
+        from nautilus_dolphin.nautilus import esof_size_gate  # type: ignore
+    return esof_size_gate
+
+
+def posture_from_engine_snapshot(snapshot: Optional[Mapping[str, Any]]) -> str:
+    """BLUE's ``engine_snapshot['posture']`` (DOLPHIN_STATE_BLUE), defaulting to APEX.
+
+    Mirrors BLUE's own default (``getattr(self, '_day_posture', 'APEX')``,
+    esf_alpha_orchestrator.py:365/613). Upper-cased for the SizingFactors contract.
+    """
+    if not isinstance(snapshot, Mapping):
+        return "APEX"
+    raw = snapshot.get("posture")
+    text = str(raw).strip() if raw is not None else ""
+    return text.upper() if text else "APEX"
+
+
+def esof_score_from_features(
+    esof_raw: Any,
+    *,
+    max_age_s: Optional[float] = None,
+) -> Optional[float]:
+    """Extract the EsoF advisory score from a raw HZ ``esof_latest`` value.
+
+    Mirrors BLUE's ``_read_esof_payload`` two-step exactly: ``parse_esof_payload(raw)``
+    (the HZ value is a raw JSON blob) then ``esof_score_from_payload`` — both BLUE's
+    OWN functions (nautilus_event_trader.py:716,729), no reimplementation. ``max_age_s``
+    mirrors BLUE's freshness gate (``ESOF_FRESHNESS_S``); ``None`` skips the staleness
+    check. Returns ``None`` when the value is missing/unparseable/stale — SizingFactors
+    then leaves ``esof_score`` unset (BLUE's ``esof_size_mult_from_score(None)`` neutral
+    path).
+    """
+    if esof_raw is None:
+        return None
+    gate = _import_esof_gate()
+    payload = gate.parse_esof_payload(esof_raw)
+    if not payload:
+        return None
+    score = gate.esof_score_from_payload(payload, max_age_s=max_age_s)
+    return None if score is None else float(score)
+
+
+@typed
+def source_live_sizing_factors(
+    *,
+    engine_snapshot: Optional[Mapping[str, Any]] = None,
+    esof_payload: Any = None,
+    esof_max_age_s: Optional[float] = None,
+) -> SizingFactors:
+    """Build live ``SizingFactors`` from BLUE's published HZ blobs.
+
+    ``posture`` and ``esof_score`` are sourced LIVE from ``engine_snapshot`` and the
+    ``esof_latest`` payload; the six organ-derived factors fall to BLUE's neutral
+    sentinels via the ``extract_live_sizing_factors`` defaults. The DARK service is
+    expected to fetch ``DOLPHIN_STATE_BLUE['engine_snapshot']`` and
+    ``DOLPHIN_FEATURES['esof_latest']`` and pass them here.
+    """
+    posture = posture_from_engine_snapshot(engine_snapshot)
+    esof_score = esof_score_from_features(esof_payload, max_age_s=esof_max_age_s)
+
+    hz_snapshot: dict[str, Any] = {"posture": posture}
+    if esof_score is not None:
+        hz_snapshot["esof_score"] = esof_score
+
+    return extract_live_sizing_factors(hz_snapshot=hz_snapshot)

prod/clean_arch/violet/shadow_journal.py

@@ -40,6 +40,14 @@ class DecisionRow(StrictModel):
     ars_score: float = Field(allow_inf_nan=False)
     bucket_idx: int = Field(ge=0, le=255)        # UInt8
     actuated: int = Field(ge=0, le=1)
+    # V3.4 full-sizing breakdown (Nullable(Float64) in DDL): NULL on the base-only
+    # path, populated when live SizingFactors drive the decision. ge=0.0 — leverage
+    # and all four multipliers are non-negative by BLUE's construction.
+    base_leverage: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
+    dc_lev_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
+    regime_size_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
+    market_ob_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
+    esof_size_mult: Optional[float] = Field(default=None, ge=0.0, allow_inf_nan=False)
 
 
 class VioletDecisionJournal:
@@ -70,6 +78,13 @@ class VioletDecisionJournal:
                 ars_score=float(decision.ars_score),
                 bucket_idx=int(decision.bucket_idx),
                 actuated=1 if decision.actuated else 0,
+                # getattr-with-None: duck-typed decisions (and the base-only path)
+                # may omit the breakdown; it stays NULL rather than raising here.
+                base_leverage=getattr(decision, "base_leverage", None),
+                dc_lev_mult=getattr(decision, "dc_lev_mult", None),
+                regime_size_mult=getattr(decision, "regime_size_mult", None),
+                market_ob_mult=getattr(decision, "market_ob_mult", None),
+                esof_size_mult=getattr(decision, "esof_size_mult", None),
             )
         except ValidationError as exc:
             self.rows_rejected += 1

prod/clean_arch/violet/shadow_live_factors.py

@@ -0,0 +1,79 @@
+"""VIOLET launcher shadow helpers for live BLUE factor sourcing.
+
+These helpers stay separate from the launcher module so they can be unit-tested
+without importing the full launcher import chain.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+LOGGER = logging.getLogger(__name__)
+
+
+def build_shadow_live_source(
+    *,
+    client_factory=None,
+    selector_factory=None,
+    source_factory=None,
+    scan_history_factory=None,
+):
+    """Create the read-only BLUE live-factor mirror for the shadow path."""
+    if client_factory is None or selector_factory is None or source_factory is None or scan_history_factory is None:
+        import hazelcast
+
+        from .alpha_wrappers import VioletAssetSelector
+        from .live_blue_source import LiveBlueScanHistory, source_live_blue_sizing_factors
+
+        client_factory = client_factory or (lambda: hazelcast.HazelcastClient(
+            cluster_name=os.environ.get("HZ_CLUSTER", "dolphin"),
+            cluster_members=[os.environ.get("HZ_HOST", "localhost:5701")],
+        ))
+        selector_factory = selector_factory or VioletAssetSelector
+        source_factory = source_factory or source_live_blue_sizing_factors
+        scan_history_factory = scan_history_factory or LiveBlueScanHistory
+
+    client = client_factory()
+    return {
+        "client": client,
+        "scan_history": scan_history_factory(),
+        "selector": selector_factory(),
+        "live_source": source_factory,
+    }
+
+
+def shadow_decision_step(
+    shadow: dict,
+    payload: dict,
+    *,
+    scan_number: int,
+    now_ns: int,
+    vel_div: float,
+    vol_ok: bool,
+) -> bool:
+    """Run one shadow decision against the live BLUE factor plane."""
+    shadow["engine"].observe(payload, scan_number)
+    live_source = shadow.get("live_source")
+    factors = None
+    if live_source is not None:
+        live_result = live_source(
+            shadow["client"],
+            scan_history=shadow["scan_history"],
+            selector=shadow["selector"],
+        )
+        shadow["last_live_source"] = live_result
+        factors = live_result.factors
+    if factors is None:
+        return False
+    decision = shadow["engine"].decide(
+        now_ns=now_ns,
+        scan_number=scan_number,
+        capital=shadow["capital"],
+        vel_div=vel_div,
+        vol_ok=vol_ok,
+        factors=factors,
+    )
+    if decision is None:
+        return False
+    return shadow["journal"].journal(decision, mono_ns=now_ns)

prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py

@@ -0,0 +1,155 @@
+"""V3.4b launcher shadow wiring — live BLUE factor plane is mandatory."""
+
+from __future__ import annotations
+
+import sys
+from types import SimpleNamespace
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+from prod.clean_arch.violet.decision_engine import ShadowDecision, SizingFactors
+from prod.clean_arch.violet.shadow_journal import VioletDecisionJournal
+
+
+def test_build_shadow_includes_live_factor_source():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    class FakeClient:
+        pass
+
+    shadow = slf.build_shadow_live_source(
+        client_factory=lambda: FakeClient(),
+        selector_factory=lambda: object(),
+        source_factory=lambda client, scan_history, selector: object(),
+        scan_history_factory=lambda: object(),
+    )
+    assert isinstance(shadow["client"], FakeClient)
+    assert shadow["live_source"] is not None
+    assert shadow["scan_history"] is not None
+    assert shadow["selector"] is not None
+
+
+def test_build_shadow_propagates_client_factory_failure():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    try:
+        slf.build_shadow_live_source(
+            client_factory=lambda: (_ for _ in ()).throw(RuntimeError("hz down")),
+            selector_factory=lambda: object(),
+            source_factory=lambda client, scan_history, selector: object(),
+            scan_history_factory=lambda: object(),
+        )
+    except RuntimeError as exc:
+        assert "hz down" in str(exc)
+    else:
+        raise AssertionError("expected live source failure")
+
+
+def test_shadow_decision_step_uses_live_factors_and_journals():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    observed = []
+    decided = []
+    journal_rows = []
+
+    class FakeEngine:
+        def observe(self, payload, scan_number):
+            observed.append((scan_number, payload["vel_div"]))
+
+        def decide(self, **kwargs):
+            decided.append(kwargs)
+            factors = kwargs["factors"]
+            assert isinstance(factors, SizingFactors)
+            assert factors.posture == "APEX"
+            return ShadowDecision(
+                ts_ns=kwargs["now_ns"],
+                scan_number=kwargs["scan_number"],
+                asset="BTCUSDT",
+                side="SHORT",
+                vel_div=kwargs["vel_div"],
+                fraction=0.2,
+                conviction_leverage=3.0,
+                notional_fraction=0.6,
+                target_exposure=41400.0,
+                ars_score=1.23,
+                bucket_idx=1,
+                actuated=True,
+                base_leverage=1.0,
+                dc_lev_mult=1.0,
+                regime_size_mult=1.0,
+                market_ob_mult=1.0,
+                esof_size_mult=1.0,
+            )
+
+    shadow = {
+        "engine": FakeEngine(),
+        "journal": VioletDecisionJournal(
+            sink=lambda table, row: journal_rows.append((table, row)),
+            session_id="sess",
+        ),
+        "capital": 69_000.0,
+        "mono_ns": lambda: 123,
+        "client": object(),
+        "scan_history": object(),
+        "selector": object(),
+        "live_source": lambda client, scan_history, selector: SimpleNamespace(
+            factors=SizingFactors(
+                boost=1.4,
+                beta=0.2,
+                mc_scale=0.5,
+                esof_score=0.42,
+                ob_median_imbalance=0.12,
+                ob_agreement_pct=0.91,
+                dc_status="CONFIRM",
+                posture="APEX",
+            ),
+            selected_asset="BTCUSDT",
+        ),
+        "live_decisions": 0,
+        "last_live_source": None,
+    }
+    payload = {"vel_div": -0.031, "vol_ok": True}
+    ok = slf.shadow_decision_step(
+        shadow,
+        payload,
+        scan_number=7,
+        now_ns=123,
+        vel_div=-0.031,
+        vol_ok=True,
+    )
+    assert ok is True
+    assert observed == [(7, -0.031)]
+    assert decided and decided[0]["factors"].dc_status == "CONFIRM"
+    assert shadow["last_live_source"].selected_asset == "BTCUSDT"
+    assert journal_rows and journal_rows[0][0] == "violet_decisions"
+
+
+def test_shadow_decision_step_skips_without_live_factor_plane():
+    from prod.clean_arch.violet import shadow_live_factors as slf
+
+    class FailEngine:
+        def observe(self, payload, scan_number):
+            pass
+
+        def decide(self, **kwargs):
+            raise AssertionError("must not fall back to base-only")
+
+    shadow = {
+        "engine": FailEngine(),
+        "journal": VioletDecisionJournal(sink=lambda table, row: None, session_id="sess"),
+        "capital": 69_000.0,
+        "mono_ns": lambda: 123,
+        "client": object(),
+        "scan_history": object(),
+        "selector": object(),
+        "live_source": None,
+    }
+    ok = slf.shadow_decision_step(
+        shadow,
+        {"vel_div": -0.031, "vol_ok": True},
+        scan_number=7,
+        now_ns=123,
+        vel_div=-0.031,
+        vol_ok=True,
+    )
+    assert ok is False

prod/clean_arch/violet/test_violet_live_blue_source.py

@@ -0,0 +1,381 @@
+from __future__ import annotations
+
+import json
+import sys
+from dataclasses import dataclass
+
+import pytest
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+import hazelcast
+
+from prod.clean_arch.violet.decision_engine import SizingFactors
+from prod.clean_arch.violet.live_blue_source import (
+    HazelcastOBProvider,
+    LiveBlueScanHistory,
+    source_live_blue_sizing_factors,
+)
+from prod.clean_arch.violet.alpha_wrappers import VioletAssetSelector
+from nautilus_dolphin.nautilus.alpha_signal_generator import AlphaSignalGenerator
+
+
+@dataclass
+class _FakeMap:
+    payloads: dict
+
+    def get(self, key):
+        return self.payloads.get(key)
+
+    def key_set(self):
+        return list(self.payloads.keys())
+
+
+class _FakeBlocking:
+    def __init__(self, payloads):
+        self._payloads = payloads
+
+    def get(self, key):
+        return self._payloads.get(key)
+
+    def key_set(self):
+        return list(self._payloads.keys())
+
+
+class _FakeClient:
+    def __init__(self, maps):
+        self._maps = maps
+
+    def get_map(self, name):
+        return type("M", (), {"blocking": lambda self2: _FakeBlocking(self._maps[name])})()
+
+
+def test_hz_ob_provider_filters_and_parses_latest_payload():
+    client = _FakeClient(
+        {
+            "DOLPHIN_FEATURES": {
+                "asset_BTCUSDT_ob": json.dumps(
+                    {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1],
+                     "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                ),
+                "asset_XRPUSDT_ob": json.dumps(
+                    {"timestamp": 2.0, "bid_notional": [-1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1],
+                     "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                ),
+                "acb_boost": json.dumps({"boost": 1.2, "beta": 0.3}),
+                "mc_forewarner_latest": json.dumps({"status": "ORANGE"}),
+                "esof_latest": json.dumps({"advisory_score": 0.4}),
+            },
+            "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "restored"})},
+        }
+    )
+    provider = HazelcastOBProvider(client)  # type: ignore[arg-type]
+    assert provider.get_assets() == ["BTCUSDT", "XRPUSDT"]
+    snap = provider.get_snapshot("BTCUSDT", 0.0)
+    assert snap is not None
+    assert snap.asset == "BTCUSDT"
+    assert snap.bid_notional.tolist() == [1.0, 2.0, 3.0, 4.0, 5.0]
+
+
+def test_source_live_blue_sizing_factors_unit(monkeypatch):
+    class FakeEngine:
+        def __init__(self, provider):
+            self.provider = provider
+        def step_live(self, assets, bar_idx):
+            assert "BTCUSDT" in assets
+        def get_market(self, ts, assets):
+            return type("M", (), {"median_imbalance": 0.12, "agreement_pct": 0.91})()
+
+    monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine)
+    history = LiveBlueScanHistory(maxlen=16, trade_direction=-1)
+    for idx, px in enumerate([100.0, 99.5, 99.0, 98.4, 97.8, 97.0, 96.2], start=1):
+        history.ingest_scan(
+            {
+                "scan_number": idx,
+                "timestamp": float(idx),
+                "vel_div": -0.031,
+                "target_asset": "BTCUSDT",
+                "assets": ["BTCUSDT"],
+                "asset_prices": [px],
+            }
+        )
+    client = _FakeClient(
+        {
+            "DOLPHIN_FEATURES": {
+                "asset_BTCUSDT_ob": json.dumps(
+                    {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1],
+                     "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                ),
+                "acb_boost": json.dumps({"boost": 1.4, "beta": 0.2}),
+                "mc_forewarner_latest": json.dumps({"status": "ORANGE"}),
+                "esof_latest": json.dumps({"advisory_score": 0.4}),
+                "latest_eigen_scan": json.dumps(
+                    {
+                        "scan_number": 8,
+                        "timestamp": 8.0,
+                        "vel_div": -0.031,
+                        "target_asset": "BTCUSDT",
+                        "assets": ["BTCUSDT"],
+                        "asset_prices": [95.5],
+                    }
+                ),
+            },
+            "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "stalker"})},
+        }
+    )
+    res = source_live_blue_sizing_factors(
+        client,
+        assets=["BTCUSDT"],
+        scan_history=history,
+        selector=VioletAssetSelector(lookback_horizon=7),
+    )
+    assert isinstance(res.factors, SizingFactors)
+    assert res.factors.posture == "STALKER"
+    assert res.factors.mc_scale == 0.5
+    assert res.factors.boost == 1.4
+    assert res.factors.beta == 0.2
+    assert res.factors.esof_score == 0.4
+    assert res.factors.ob_median_imbalance == 0.12
+    assert res.factors.ob_agreement_pct == 0.91
+    assert res.factors.dc_status == "CONFIRM"
+    assert res.selected_asset == "BTCUSDT"
+
+
+def test_source_live_blue_sizing_factors_preserves_skip_contradict(monkeypatch):
+    class FakeEngine:
+        def __init__(self, provider):
+            self.provider = provider
+        def step_live(self, assets, bar_idx):
+            pass
+        def get_market(self, ts, assets):
+            return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})()
+
+    monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine)
+    history = LiveBlueScanHistory(maxlen=16, trade_direction=-1)
+    for idx, px in enumerate([100.0, 100.5, 101.0, 101.6, 102.2, 102.9, 103.6], start=1):
+        history.ingest_scan(
+            {
+                "scan_number": idx,
+                "timestamp": float(idx),
+                "vel_div": -0.031,
+                "target_asset": "BTCUSDT",
+                "assets": ["BTCUSDT"],
+                "asset_prices": [px],
+            }
+        )
+    client = _FakeClient(
+        {
+            "DOLPHIN_FEATURES": {
+                "asset_BTCUSDT_ob": json.dumps(
+                    {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1],
+                     "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                ),
+                "acb_boost": json.dumps({"boost": 1.4, "beta": 0.2}),
+                "mc_forewarner_latest": json.dumps({"status": "GREEN"}),
+                "esof_latest": json.dumps({"advisory_score": 0.4}),
+                "latest_eigen_scan": json.dumps(
+                    {
+                        "scan_number": 8,
+                        "timestamp": 8.0,
+                        "vel_div": -0.031,
+                        "target_asset": "BTCUSDT",
+                        "assets": ["BTCUSDT"],
+                        "asset_prices": [104.2],
+                    }
+                ),
+            },
+            "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})},
+        }
+    )
+    res = source_live_blue_sizing_factors(
+        client,
+        assets=["BTCUSDT"],
+        scan_history=history,
+        selector=VioletAssetSelector(lookback_horizon=7),
+    )
+    assert res.factors.dc_status == "SKIP_CONTRADICT"
+
+
+def test_live_blue_sequence_matches_blue_selector_and_dc_at_each_step(monkeypatch):
+    class FakeEngine:
+        def __init__(self, provider):
+            self.provider = provider
+        def step_live(self, assets, bar_idx):
+            pass
+        def get_market(self, ts, assets):
+            return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})()
+
+    monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine)
+    selector = VioletAssetSelector(lookback_horizon=7)
+    history = LiveBlueScanHistory(maxlen=16, trade_direction=-1)
+    signal_gen = AlphaSignalGenerator()
+
+    scans = [
+        {
+            "scan_number": 1,
+            "timestamp": 1.0,
+            "vel_div": -0.010,
+            "assets": ["BTCUSDT", "ETHUSDT"],
+            "asset_prices": [100.0, 200.0],
+        },
+        {
+            "scan_number": 2,
+            "timestamp": 2.0,
+            "vel_div": -0.031,
+            "assets": ["BTCUSDT", "ETHUSDT"],
+            "asset_prices": [99.0, 198.0],
+        },
+        {
+            "scan_number": 3,
+            "timestamp": 3.0,
+            "vel_div": -0.041,
+            "assets": ["BTCUSDT", "ETHUSDT"],
+            "asset_prices": [98.0, 196.0],
+        },
+        {
+            "scan_number": 4,
+            "timestamp": 4.0,
+            "vel_div": -0.031,
+            "assets": ["BTCUSDT", "ETHUSDT"],
+            "asset_prices": [97.0, 194.0],
+        },
+    ]
+
+    for idx, scan in enumerate(scans, start=1):
+        client = _FakeClient(
+            {
+                "DOLPHIN_FEATURES": {
+                    "asset_BTCUSDT_ob": json.dumps(
+                        {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1],
+                         "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                    ),
+                    "asset_ETHUSDT_ob": json.dumps(
+                        {"timestamp": 1.0, "bid_notional": [2, 3, 4, 5, 6], "ask_notional": [6, 5, 4, 3, 2],
+                         "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                    ),
+                    "acb_boost": json.dumps({"boost": 1.0, "beta": 0.0}),
+                    "mc_forewarner_latest": json.dumps({"status": "GREEN"}),
+                    "esof_latest": json.dumps({"advisory_score": 0.3}),
+                    "latest_eigen_scan": json.dumps({**scan, "target_asset": "BTCUSDT"}),
+                },
+                "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})},
+            }
+        )
+        res = source_live_blue_sizing_factors(
+            client,
+            assets=["BTCUSDT", "ETHUSDT"],
+            scan_history=history,
+            selector=selector,
+        )
+
+        # BLUE selector parity
+        market = history.market_data(selector.lookback)
+        expected_pick = selector.pick(market, regime_direction=-1)
+        expected_asset = expected_pick.asset if expected_pick is not None else "BTCUSDT"
+        assert res.selected_asset == expected_asset
+
+        # BLUE signal parity
+        expected_signal = signal_gen.generate(
+            vel_div=float(scan["vel_div"]),
+            vel_div_history=None,
+            asset_price_history=history.price_history(expected_asset),
+            trade_direction=-1,
+            asset=expected_asset,
+            current_timestamp=float(scan["timestamp"]),
+        )
+        assert res.factors.dc_status == expected_signal.dc_status
+
+
+def test_live_blue_sequence_rejects_anomalous_values_without_poisoning_history(monkeypatch):
+    class FakeEngine:
+        def __init__(self, provider):
+            self.provider = provider
+        def step_live(self, assets, bar_idx):
+            pass
+        def get_market(self, ts, assets):
+            return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})()
+
+    monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine)
+    history = LiveBlueScanHistory(maxlen=16, trade_direction=-1)
+    selector = VioletAssetSelector(lookback_horizon=7)
+    scan = {
+        "scan_number": 99,
+        "timestamp": 99.0,
+        "vel_div": -0.031,
+        "assets": ["BTCUSDT", "ETHUSDT"],
+        "asset_prices": [float("nan"), -1.0],
+        "target_asset": "BTCUSDT",
+    }
+    client = _FakeClient(
+        {
+            "DOLPHIN_FEATURES": {
+                "asset_BTCUSDT_ob": json.dumps(
+                    {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1],
+                     "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                ),
+                "acb_boost": json.dumps({"boost": 1.0, "beta": 0.0}),
+                "mc_forewarner_latest": json.dumps({"status": "GREEN"}),
+                "esof_latest": json.dumps({"advisory_score": 0.3}),
+                "latest_eigen_scan": json.dumps(scan),
+            },
+            "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": "APEX"})},
+        }
+    )
+    res = source_live_blue_sizing_factors(
+        client,
+        assets=["BTCUSDT", "ETHUSDT"],
+        scan_history=history,
+        selector=selector,
+    )
+    assert res.selected_asset == "BTCUSDT"
+    assert res.factors.dc_status == "NONE"
+    assert history.price_history("BTCUSDT") == []
+    assert history.price_history("ETHUSDT") == []
+
+
+def test_source_live_blue_sizing_factors_handles_anomalies(monkeypatch):
+    class FakeEngine:
+        def __init__(self, provider):
+            self.provider = provider
+        def step_live(self, assets, bar_idx):
+            raise RuntimeError("boom")
+        def get_market(self, ts, assets):
+            return type("M", (), {"median_imbalance": 0.0, "agreement_pct": 0.0})()
+
+    monkeypatch.setattr("prod.clean_arch.violet.live_blue_source.OBFeatureEngine", FakeEngine)
+    client = _FakeClient(
+        {
+            "DOLPHIN_FEATURES": {
+                "asset_BTCUSDT_ob": json.dumps(
+                    {"timestamp": 1.0, "bid_notional": [1, 2, 3, 4, 5], "ask_notional": [5, 4, 3, 2, 1],
+                     "bid_depth": [1, 1, 1, 1, 1], "ask_depth": [1, 1, 1, 1, 1]}
+                ),
+                "acb_boost": json.dumps({"boost": -9, "beta": "bad"}),
+                "mc_forewarner_latest": json.dumps({"status": "GREEN"}),
+                "esof_latest": "not json",
+                "latest_eigen_scan": json.dumps({"target_asset": "BTCUSDT", "assets": ["BTCUSDT"], "asset_prices": [0]}),
+            },
+            "DOLPHIN_STATE_BLUE": {"latest_nautilus": json.dumps({"posture": ""})},
+        }
+    )
+    res = source_live_blue_sizing_factors(client, assets=["BTCUSDT"])
+    assert res.factors.posture == "APEX"
+    assert res.factors.mc_scale == 1.0
+    assert res.factors.boost == 0.0
+    assert res.factors.beta == 0.0
+    assert res.factors.esof_score is None
+    assert res.factors.ob_median_imbalance is None
+    assert res.factors.ob_agreement_pct is None
+    assert res.factors.dc_status == "NONE"
+    assert res.selected_asset == "BTCUSDT"
+
+
+def test_live_hz_smoke_reads_current_state():
+    client = hazelcast.HazelcastClient(cluster_name="dolphin", cluster_members=["localhost:5701"])
+    try:
+        res = source_live_blue_sizing_factors(client, assets=["BTCUSDT", "ETHUSDT", "XRPUSDT"])
+    finally:
+        client.shutdown()
+    assert isinstance(res.factors, SizingFactors)
+    assert res.factors.posture in {"APEX", "STALKER", "RESTORED", "TURTLE", "HIBERNATE"}
+    assert res.factors.mc_scale in {0.5, 1.0}

prod/clean_arch/violet/test_violet_live_factor_source.py

@@ -0,0 +1,85 @@
+"""V3.4b: live_factor_source — SizingFactors from BLUE's published HZ blobs.
+
+Validates the field-path findings in
+prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md: posture + esof_score are
+sourced LIVE from engine_snapshot / the esof_latest payload, the six organ-derived
+factors fall to BLUE's neutral sentinels.
+"""
+
+from __future__ import annotations
+
+import sys
+
+import pytest
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+from prod.clean_arch.violet.decision_engine import SizingFactors
+from prod.clean_arch.violet.live_factor_source import (
+    ORGAN_DERIVED_FACTORS,
+    esof_score_from_features,
+    posture_from_engine_snapshot,
+    source_live_sizing_factors,
+)
+
+
+def test_posture_sourced_and_upper_cased():
+    assert posture_from_engine_snapshot({"posture": "STALKER"}) == "STALKER"
+    assert posture_from_engine_snapshot({"posture": "restored"}) == "RESTORED"
+
+
+def test_posture_defaults_to_apex_like_blue():
+    assert posture_from_engine_snapshot(None) == "APEX"
+    assert posture_from_engine_snapshot({}) == "APEX"
+    assert posture_from_engine_snapshot({"posture": ""}) == "APEX"
+    assert posture_from_engine_snapshot({"posture": None}) == "APEX"
+
+
+def test_esof_score_parsed_from_dict_payload():
+    # max_age_s=None skips staleness; advisory_score preferred over score.
+    assert esof_score_from_features({"advisory_score": 0.5}) == 0.5
+    assert esof_score_from_features({"score": -0.1}) == pytest.approx(-0.1)
+
+
+def test_esof_score_parsed_from_raw_json_string():
+    # The HZ value is a raw JSON blob — BLUE's parse_esof_payload handles it.
+    assert esof_score_from_features('{"advisory_score": 0.25}') == 0.25
+
+
+def test_esof_score_none_when_missing_or_unparseable():
+    assert esof_score_from_features(None) is None
+    assert esof_score_from_features("not json") is None
+    assert esof_score_from_features({}) is None        # no advisory_score/score key
+
+
+def test_esof_staleness_gate_honored_when_max_age_supplied():
+    # A payload with an ancient timestamp is stale → None when max_age_s is set.
+    stale = {"advisory_score": 0.5, "unix": 0.0}       # 1970 → very old
+    assert esof_score_from_features(stale, max_age_s=30.0) is None
+    # …but with no freshness gate (None) the score still comes through.
+    assert esof_score_from_features(stale, max_age_s=None) == 0.5
+
+
+def test_source_live_factors_posture_and_esof_live_rest_neutral():
+    factors = source_live_sizing_factors(
+        engine_snapshot={"posture": "RESTORED", "capital": 71591.1},
+        esof_payload={"advisory_score": 0.42},
+    )
+    assert isinstance(factors, SizingFactors)
+    assert factors.posture == "RESTORED"
+    assert factors.esof_score == 0.42
+    # the six organ-derived factors at BLUE's own neutral sentinels (V3.4c will source)
+    assert factors.boost == 1.0 and factors.beta == 0.0 and factors.mc_scale == 1.0
+    assert factors.ob_median_imbalance is None and factors.ob_agreement_pct is None
+    assert factors.dc_status == "NONE"
+
+
+def test_source_live_factors_all_neutral_when_no_blobs():
+    assert source_live_sizing_factors() == SizingFactors()
+
+
+def test_organ_derived_factor_set_is_the_documented_six():
+    assert set(ORGAN_DERIVED_FACTORS) == {
+        "boost", "beta", "mc_scale",
+        "ob_median_imbalance", "ob_agreement_pct", "dc_status",
+    }

prod/clean_arch/violet/test_violet_shadow_journal.py

@@ -71,3 +71,47 @@ def test_negative_exposure_rejected():
     )
     assert j.journal(bad, mono_ns=1) is False
     assert j.rows_rejected == 1
+
+
+_BREAKDOWN_COLS = (
+    "base_leverage", "dc_lev_mult", "regime_size_mult",
+    "market_ob_mult", "esof_size_mult",
+)
+
+
+def test_full_factor_breakdown_round_trips_into_row():
+    captured = []
+    j = VioletDecisionJournal(sink=lambda t, r: captured.append(r), session_id="s")
+    dec = _decision(
+        base_leverage=4.0, dc_lev_mult=1.5, regime_size_mult=1.2,
+        market_ob_mult=1.4, esof_size_mult=0.95,
+    )
+    assert j.journal(dec, mono_ns=1) and j.rows_emitted == 1
+    row = captured[0]
+    assert set(row.keys()) == _ddl_columns()        # parity holds with new columns
+    assert row["base_leverage"] == 4.0 and row["dc_lev_mult"] == 1.5
+    assert row["regime_size_mult"] == 1.2 and row["market_ob_mult"] == 1.4
+    assert row["esof_size_mult"] == 0.95
+
+
+def test_base_only_path_leaves_breakdown_null():
+    captured = []
+    j = VioletDecisionJournal(sink=lambda t, r: captured.append(r), session_id="s")
+    assert j.journal(_decision(), mono_ns=1)        # no breakdown supplied
+    row = captured[0]
+    assert all(row[c] is None for c in _BREAKDOWN_COLS)
+
+
+def test_negative_breakdown_multiplier_rejected_at_source():
+    j = VioletDecisionJournal(sink=lambda t, r: None, session_id="s")
+    # ShadowDecision itself guards ge=0.0, so build a duck-typed decision that
+    # smuggles a negative multiplier past the engine to prove the row guard catches it.
+    bad = SimpleNamespace(
+        scan_number=1, asset="BTCUSDT", side="SHORT", vel_div=-0.2,
+        fraction=0.2, conviction_leverage=9.0, notional_fraction=1.8,
+        target_exposure=1.0, ars_score=1.0, bucket_idx=1, actuated=True,
+        base_leverage=4.0, dc_lev_mult=-0.5, regime_size_mult=1.0,
+        market_ob_mult=1.0, esof_size_mult=1.0,
+    )
+    assert j.journal(bad, mono_ns=1) is False
+    assert j.rows_rejected == 1

prod/clean_arch/violet/test_violet_trade_slot_compare.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from prod.clean_arch.violet.trade_slot_compare import (
+    compare_trade_slot_granularity,
+)
+
+
+def test_compare_trade_slot_granularity_collapses_and_matches():
+    decisions = [
+        {
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "scan_number": 10,
+            "ts": 1_000,
+            "actuated": True,
+            "conviction_leverage": 3.0,
+            "target_exposure": 30.0,
+        },
+        {
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "scan_number": 11,
+            "ts": 2_000,
+            "actuated": True,
+            "conviction_leverage": 4.0,
+            "target_exposure": 40.0,
+        },
+        {
+            "asset": "BTCUSDT",
+            "side": "LONG",
+            "scan_number": 16,
+            "ts": 4_000,
+            "actuated": True,
+            "conviction_leverage": 2.0,
+            "target_exposure": 20.0,
+        },
+    ]
+    trades = [
+        {
+            "trade_id": "t-1",
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "ts": 900,
+            "bars_held": 1,
+            "net_pnl": 1.5,
+            "reason": "OPEN",
+        },
+        {
+            "trade_id": "t-1",
+            "asset": "BTCUSDT",
+            "side": "SHORT",
+            "ts": 2_500,
+            "bars_held": 2,
+            "net_pnl": 4.5,
+            "reason": "EXIT",
+        },
+        {
+            "trade_id": "t-2",
+            "asset": "BTCUSDT",
+            "side": "LONG",
+            "ts": 3_900,
+            "bars_held": 1,
+            "net_pnl": -0.25,
+            "reason": "EXIT",
+        },
+    ]
+
+    result = compare_trade_slot_granularity(decisions, trades)
+
+    assert len(result.decision_episodes) == 2
+    assert len(result.trade_episodes) == 2
+    assert len(result.matches) == 2
+    assert not result.decision_only
+    assert not result.trade_only
+
+    short_match = result.matches[0]
+    assert short_match.asset == "BTCUSDT"
+    assert short_match.side == "SHORT"
+    assert short_match.decision_episode.first_scan_number == 10
+    assert short_match.decision_episode.last_scan_number == 11
+    assert short_match.decision_episode.row_count == 2
+    assert short_match.trade_episode.trade_id == "t-1"
+    assert short_match.trade_episode.row_count == 2
+    assert short_match.trade_episode.terminal_reason == "EXIT"
+    assert short_match.trade_episode.net_pnl == 4.5
+    assert short_match.start_gap_ms == 100
+    assert short_match.end_gap_ms == 500
+    assert short_match.row_gap == 0
+    assert short_match.bars_gap == 0
+
+
+def test_compare_trade_slot_granularity_splits_on_scan_gap_and_ignores_bad_rows():
+    decisions = [
+        {
+            "asset": "ETHUSDT",
+            "side": "SHORT",
+            "scan_number": 1,
+            "ts": 10,
+            "actuated": True,
+            "conviction_leverage": 1.0,
+            "target_exposure": 10.0,
+        },
+        {
+            "asset": "ETHUSDT",
+            "side": "SHORT",
+            "scan_number": 2,
+            "ts": 20,
+            "actuated": True,
+            "conviction_leverage": 1.1,
+            "target_exposure": 11.0,
+        },
+        {
+            "asset": "ETHUSDT",
+            "side": "SHORT",
+            "scan_number": 8,
+            "ts": 80,
+            "actuated": True,
+            "conviction_leverage": 1.2,
+            "target_exposure": 12.0,
+        },
+        {
+            "asset": None,
+            "side": "SHORT",
+            "scan_number": "bad",
+            "ts": 90,
+            "actuated": True,
+        },
+    ]
+    trades = [
+        {"trade_id": "x-1", "asset": "ETHUSDT", "side": "SHORT", "ts": 15, "reason": "EXIT"},
+        {"trade_id": "x-2", "asset": "ETHUSDT", "side": "SHORT", "ts": 99, "reason": "EXIT"},
+    ]
+
+    result = compare_trade_slot_granularity(decisions, trades)
+
+    assert len(result.decision_episodes) == 2
+    assert [ep.row_count for ep in result.decision_episodes] == [2, 1]
+    assert len(result.trade_episodes) == 2
+    assert len(result.matches) == 2
+    assert [m.trade_episode.trade_id for m in result.matches] == ["x-1", "x-2"]
+    assert not result.decision_only
+    assert not result.trade_only
+
+
+def test_compare_trade_slot_granularity_handles_empty_input():
+    result = compare_trade_slot_granularity([], [])
+    assert result.decision_episodes == []
+    assert result.trade_episodes == []
+    assert result.matches == []
+    assert result.decision_only == []
+    assert result.trade_only == []

prod/clean_arch/violet/trade_slot_compare.py

@@ -0,0 +1,309 @@
+"""VIOLET trade/slot comparison harness.
+
+This is the missing V3 comparison unit from the main spec: collapse shadow
+decisions into episode-sized runs, collapse raw trade rows into terminal trade
+episodes, and compare them without requiring live execution wiring.
+
+VIOLET-only. BLUE is untouched.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Iterable, Mapping
+from typing import Any, Optional
+
+from pydantic import Field
+
+from .domain import StrictModel, Symbol, typed
+
+
+def _coerce_int(value: Any, default: Optional[int] = None) -> Optional[int]:
+    try:
+        if value is None:
+            return default
+        out = int(value)
+        return out if out >= 0 else default
+    except (TypeError, ValueError):
+        return default
+
+
+def _coerce_float(value: Any, default: Optional[float] = None) -> Optional[float]:
+    try:
+        if value is None:
+            return default
+        out = float(value)
+        return out if out == out and out not in (float("inf"), float("-inf")) else default
+    except (TypeError, ValueError):
+        return default
+
+
+def _coerce_text(value: Any, default: str = "") -> str:
+    if value is None:
+        return default
+    text = str(value).strip()
+    return text if text else default
+
+
+def _row_asset(row: Mapping[str, Any]) -> str:
+    return _coerce_text(row.get("asset") or row.get("symbol") or row.get("instrument")).upper()
+
+
+def _row_side(row: Mapping[str, Any]) -> str:
+    return _coerce_text(row.get("side") or row.get("direction") or row.get("trade_side")).upper()
+
+
+def _row_scan_number(row: Mapping[str, Any]) -> Optional[int]:
+    return _coerce_int(row.get("scan_number") or row.get("scan") or row.get("scan_idx"))
+
+
+def _row_ts_ms(row: Mapping[str, Any]) -> Optional[int]:
+    candidates = (
+        row.get("ts"),
+        row.get("ts_ms"),
+        row.get("timestamp"),
+        row.get("exit_ts"),
+        row.get("entry_ts"),
+        row.get("mono_ns"),
+    )
+    for value in candidates:
+        ts = _coerce_int(value)
+        if ts is not None:
+            return ts if value is None or value != row.get("mono_ns") else ts // 1_000_000
+    return None
+
+
+def _row_trade_id(row: Mapping[str, Any]) -> str:
+    return _coerce_text(
+        row.get("trade_id") or row.get("id") or row.get("slot_id") or row.get("episode_id"),
+    )
+
+
+class DecisionEpisode(StrictModel):
+    asset: Symbol
+    side: str = Field(min_length=1, max_length=16)
+    first_scan_number: int = Field(ge=0)
+    last_scan_number: int = Field(ge=0)
+    first_ts_ms: int = Field(ge=0)
+    last_ts_ms: int = Field(ge=0)
+    row_count: int = Field(ge=1)
+    actuated_count: int = Field(ge=0)
+    max_conviction_leverage: float = Field(ge=0.0, allow_inf_nan=False)
+    last_target_exposure: float = Field(ge=0.0, allow_inf_nan=False)
+
+
+class TradeEpisode(StrictModel):
+    trade_id: str = Field(min_length=1)
+    asset: Symbol
+    side: str = Field(min_length=1, max_length=16)
+    entry_ts_ms: int = Field(ge=0)
+    exit_ts_ms: int = Field(ge=0)
+    row_count: int = Field(ge=1)
+    bars_held: Optional[int] = Field(default=None, ge=0)
+    net_pnl: Optional[float] = Field(default=None, allow_inf_nan=False)
+    terminal_reason: Optional[str] = None
+
+
+class EpisodeMatch(StrictModel):
+    asset: Symbol
+    side: str = Field(min_length=1, max_length=16)
+    decision_episode: DecisionEpisode
+    trade_episode: TradeEpisode
+    start_gap_ms: int = Field(ge=0)
+    end_gap_ms: int = Field(ge=0)
+    row_gap: int = Field(ge=0)
+    bars_gap: Optional[int] = Field(default=None, ge=0)
+
+
+class TradeSlotComparison(StrictModel):
+    decision_episodes: list[DecisionEpisode]
+    trade_episodes: list[TradeEpisode]
+    matches: list[EpisodeMatch]
+    decision_only: list[DecisionEpisode]
+    trade_only: list[TradeEpisode]
+
+
+def _collapse_decision_rows(
+    rows: Iterable[Mapping[str, Any]],
+    *,
+    max_scan_gap: int = 1,
+) -> list[DecisionEpisode]:
+    grouped: dict[tuple[str, str], list[Mapping[str, Any]]] = defaultdict(list)
+    for row in rows:
+        asset = _row_asset(row)
+        side = _row_side(row)
+        scan_number = _row_scan_number(row)
+        if not asset or not side or scan_number is None:
+            continue
+        grouped[(asset, side)].append(row)
+
+    episodes: list[DecisionEpisode] = []
+    for (asset, side), bucket in grouped.items():
+        bucket = sorted(
+            bucket,
+            key=lambda row: (
+                _row_scan_number(row) or 0,
+                _row_ts_ms(row) or 0,
+            ),
+        )
+        current: list[Mapping[str, Any]] = []
+        prev_scan: Optional[int] = None
+        for row in bucket:
+            scan_number = _row_scan_number(row)
+            if scan_number is None:
+                continue
+            if current and prev_scan is not None and scan_number > prev_scan + max_scan_gap:
+                episodes.append(_decision_episode_from_rows(asset, side, current))
+                current = []
+            current.append(row)
+            prev_scan = scan_number
+        if current:
+            episodes.append(_decision_episode_from_rows(asset, side, current))
+
+    return sorted(episodes, key=lambda ep: (ep.first_ts_ms, ep.asset, ep.side, ep.first_scan_number))
+
+
+def _decision_episode_from_rows(
+    asset: str,
+    side: str,
+    rows: list[Mapping[str, Any]],
+) -> DecisionEpisode:
+    scans = [sn for sn in (_row_scan_number(r) for r in rows) if sn is not None]
+    times = [ts for ts in (_row_ts_ms(r) for r in rows) if ts is not None]
+    conv = [
+        value for value in (_coerce_float(r.get("conviction_leverage"), None) for r in rows)
+        if value is not None
+    ]
+    exposure = [
+        value for value in (_coerce_float(r.get("target_exposure"), None) for r in rows)
+        if value is not None
+    ]
+    actuated = sum(1 for r in rows if bool(r.get("actuated")))
+    return DecisionEpisode(
+        asset=asset,
+        side=side,
+        first_scan_number=min(scans),
+        last_scan_number=max(scans),
+        first_ts_ms=min(times) if times else 0,
+        last_ts_ms=max(times) if times else 0,
+        row_count=len(rows),
+        actuated_count=actuated,
+        max_conviction_leverage=max(conv) if conv else 0.0,
+        last_target_exposure=exposure[-1] if exposure else 0.0,
+    )
+
+
+def _collapse_trade_rows(rows: Iterable[Mapping[str, Any]]) -> list[TradeEpisode]:
+    grouped: dict[str, list[Mapping[str, Any]]] = defaultdict(list)
+    for row in rows:
+        trade_id = _row_trade_id(row)
+        asset = _row_asset(row)
+        side = _row_side(row)
+        if not trade_id or not asset or not side:
+            continue
+        grouped[trade_id].append(row)
+
+    episodes: list[TradeEpisode] = []
+    for trade_id, bucket in grouped.items():
+        bucket = sorted(bucket, key=lambda row: (_row_ts_ms(row) or 0, _coerce_int(row.get("scan_number")) or 0))
+        first = bucket[0]
+        last = bucket[-1]
+        entry_ts = _row_ts_ms(first) or 0
+        exit_ts = _row_ts_ms(last) or entry_ts
+        bars = None
+        for row in reversed(bucket):
+            bars = _coerce_int(row.get("bars_held"))
+            if bars is not None:
+                break
+        net_pnl = None
+        for row in reversed(bucket):
+            net_pnl = _coerce_float(row.get("net_pnl") or row.get("pnl"), None)
+            if net_pnl is not None:
+                break
+        reason = None
+        for row in reversed(bucket):
+            reason = _coerce_text(row.get("reason") or row.get("exit_reason"), "")
+            if reason:
+                break
+        episodes.append(
+            TradeEpisode(
+                trade_id=trade_id,
+                asset=_row_asset(first),
+                side=_row_side(first),
+                entry_ts_ms=entry_ts,
+                exit_ts_ms=exit_ts,
+                row_count=len(bucket),
+                bars_held=bars,
+                net_pnl=net_pnl,
+                terminal_reason=reason or None,
+            )
+        )
+
+    return sorted(episodes, key=lambda ep: (ep.entry_ts_ms, ep.asset, ep.side, ep.trade_id))
+
+
+def _match_episodes(
+    decisions: list[DecisionEpisode],
+    trades: list[TradeEpisode],
+) -> tuple[list[EpisodeMatch], list[DecisionEpisode], list[TradeEpisode]]:
+    by_key: dict[tuple[str, str], list[TradeEpisode]] = defaultdict(list)
+    for trade in trades:
+        by_key[(trade.asset, trade.side)].append(trade)
+    for bucket in by_key.values():
+        bucket.sort(key=lambda ep: (ep.entry_ts_ms, ep.exit_ts_ms, ep.trade_id))
+
+    matches: list[EpisodeMatch] = []
+    decision_only: list[DecisionEpisode] = []
+    used_trade_ids: set[str] = set()
+
+    for decision in decisions:
+        bucket = by_key.get((decision.asset, decision.side), [])
+        candidate = None
+        for trade in bucket:
+            if trade.trade_id in used_trade_ids:
+                continue
+            candidate = trade
+            break
+        if candidate is None:
+            decision_only.append(decision)
+            continue
+        used_trade_ids.add(candidate.trade_id)
+        matches.append(
+            EpisodeMatch(
+                asset=decision.asset,
+                side=decision.side,
+                decision_episode=decision,
+                trade_episode=candidate,
+                start_gap_ms=abs(decision.first_ts_ms - candidate.entry_ts_ms),
+                end_gap_ms=abs(decision.last_ts_ms - candidate.exit_ts_ms),
+                row_gap=abs(decision.row_count - candidate.row_count),
+                bars_gap=(
+                    abs(decision.row_count - candidate.bars_held)
+                    if candidate.bars_held is not None
+                    else None
+                ),
+            )
+        )
+
+    trade_only = [trade for trade in trades if trade.trade_id not in used_trade_ids]
+    return matches, decision_only, trade_only
+
+
+@typed
+def compare_trade_slot_granularity(
+    decision_rows: Iterable[Mapping[str, Any]],
+    trade_rows: Iterable[Mapping[str, Any]],
+    *,
+    max_scan_gap: int = 1,
+) -> TradeSlotComparison:
+    """Collapse both surfaces to episodes and compare them at slot granularity."""
+    decisions = _collapse_decision_rows(decision_rows, max_scan_gap=max_scan_gap)
+    trades = _collapse_trade_rows(trade_rows)
+    matches, decision_only, trade_only = _match_episodes(decisions, trades)
+    return TradeSlotComparison(
+        decision_episodes=decisions,
+        trade_episodes=trades,
+        matches=matches,
+        decision_only=decision_only,
+        trade_only=trade_only,
+    )

prod/clickhouse/violet/22_violet_decisions.sql

@@ -19,7 +19,16 @@ CREATE TABLE IF NOT EXISTS dolphin_violet.violet_decisions
     `target_exposure`     Float64,
     `ars_score`           Float64,
     `bucket_idx`          UInt8,
-    `actuated`            UInt8
+    `actuated`            UInt8,
+    -- V3.4 full-sizing breakdown: conviction = base × dc × regime(ACB) × ob × esof,
+    -- capped @9. NULL on the legacy base-only path (no live SizingFactors); populated
+    -- when the launcher feeds live factor planes. Additive columns — on a pre-existing
+    -- table apply the matching `ALTER TABLE ... ADD COLUMN IF NOT EXISTS` instead.
+    `base_leverage`       Nullable(Float64),
+    `dc_lev_mult`         Nullable(Float64),
+    `regime_size_mult`    Nullable(Float64),
+    `market_ob_mult`      Nullable(Float64),
+    `esof_size_mult`      Nullable(Float64)
 )
 ENGINE = MergeTree
 ORDER BY (asset, ts)

prod/docs/RECENT_VIOLET_34C_a632c59.md

@@ -0,0 +1,165 @@
+# RECENT_VIOLET_34C_a632c59
+
+## What This Work Was
+
+This change completed the first V3.4c VIOLET-side mirror for BLUE live-factor
+inputs.
+
+The goal was not to modify BLUE. The goal was to make VIOLET read the same live,
+published BLUE surfaces and reconstruct the same intermediate factors BLUE would
+see, without changing BLUE code, schemas, or state layout.
+
+## Scope
+
+This work stayed inside VIOLET and added a read-only live-source adapter plus
+tests:
+
+- `prod/clean_arch/violet/live_blue_source.py`
+- `prod/clean_arch/violet/test_violet_live_blue_source.py`
+
+It also reused the existing V3.4b live-factor helpers:
+
+- `prod/clean_arch/violet/live_factor_source.py`
+- `prod/clean_arch/violet/live_factors.py`
+- `prod/clean_arch/violet/alpha_wrappers.py`
+
+## Why This Was Needed
+
+The earlier V3.4b adapter could source only the BLUE-published pieces that were
+already directly available in Hazelcast:
+
+- `posture`
+- `esof_score`
+
+The remaining sizing inputs were not flat HZ scalars. They had to be mirrored
+from the same BLUE inputs and kernels that generate them:
+
+- `boost` / `beta` from the ACB output
+- `mc_scale` from MC-Forewarner status
+- `ob_median_imbalance` / `ob_agreement_pct` from live OB data
+- `dc_status` from the signal generator
+
+The V3.4c step is the read-only, VIOLET-side reconstruction of those live
+factors.
+
+## What Was Added
+
+### 1. Live BLUE source adapter
+
+`live_blue_source.py` now:
+
+- reads `DOLPHIN_STATE_BLUE.latest_nautilus` / `engine_snapshot`
+- reads `DOLPHIN_FEATURES.esof_latest` / `esof_advisor_latest`
+- reads `DOLPHIN_FEATURES.acb_boost`
+- reads `DOLPHIN_FEATURES.mc_forewarner_latest`
+- reads live OB shard maps from `DOLPHIN_FEATURES.asset_*_ob`
+- reconstructs `dc_status` from the published scan stream
+
+The module stays read-only. It does not write Hazelcast. It does not call BLUE
+internals for mutation. It only mirrors what BLUE already published.
+
+### 2. Stateful scan replay for DC
+
+`LiveBlueScanHistory` was added to keep a per-asset price history on the VIOLET
+side. This is needed because BLUE’s `dc_status` is derived from the live scan
+sequence and a short price history, not from a single HZ scalar.
+
+The adapter now:
+
+- ingests each `latest_eigen_scan`
+- keeps the asset histories in memory
+- uses BLUE’s own `AlphaSignalGenerator`
+- produces the same DC status labels that BLUE would emit
+
+### 3. Read-only OB mirror
+
+`HazelcastOBProvider` was added so VIOLET can feed BLUE’s own
+`OBFeatureEngine` from the live `asset_*_ob` entries already published in HZ.
+
+That lets the VIOLET path derive:
+
+- `ob_median_imbalance`
+- `ob_agreement_pct`
+
+without inventing a new OB schema or mutating BLUE.
+
+### 4. Asset selection parity
+
+The adapter now uses `VioletAssetSelector` on the replayed scan history so the
+selected asset is not guessed from the current scan payload alone.
+
+That matters because the exact factor sequence must track the same information
+BLUE would have at that point in the scan stream.
+
+## Exactness Rules Followed
+
+The implementation was kept conservative:
+
+- no BLUE file edits
+- no BLUE schema changes
+- no new HZ writers
+- no invented factor names
+- no silent fallback to fake live values when the live source exists
+
+If input is malformed, the adapter rejects or neutralizes it instead of
+poisoning the history.
+
+Examples of handled anomalies:
+
+- missing `assets`
+- non-finite prices
+- negative prices
+- malformed JSON payloads
+- missing `posture`
+- missing `esof` payloads
+- broken OB payloads
+
+## Tests Added
+
+The new test file covers three layers:
+
+### Unit tests
+
+- OB shard parsing from HZ payloads
+- neutral handling for malformed ACB / ESOF / MC payloads
+- scan replay ingestion
+- DC preservation for `CONFIRM`
+- DC preservation for `SKIP_CONTRADICT`
+
+### Sequence parity tests
+
+The new sequence test walks multiple scan events and checks that VIOLET tracks:
+
+- the BLUE asset selector output
+- the BLUE signal-generator `dc_status`
+
+at each step in the replayed scan history.
+
+### End-to-end smoke
+
+A live Hazelcast smoke test reads the current cluster state and verifies the
+adapter can build a typed `SizingFactors` object from the live BLUE surfaces.
+
+## Result
+
+The V3.4c mirror now reconstructs the full live factor plane on the VIOLET
+side, read-only, with parity-style coverage around the intermediate factor
+computation.
+
+## Verification
+
+Commit:
+
+- `a632c59` — `VIOLET V3.4c: read-only BLUE live source parity`
+
+Tests run:
+
+- `prod/clean_arch/violet/test_violet_live_blue_source.py`
+- `prod/clean_arch/violet/test_violet_live_factor_source.py`
+- `prod/clean_arch/violet/test_violet_live_factors.py`
+
+Observed result:
+
+- V3.4c source tests passed
+- existing V3.4b live-factor tests passed
+

prod/docs/RECENT_VIOLET_34D_fb34431.md

@@ -0,0 +1,102 @@
+# RECENT_VIOLET_34D_fb34431
+
+## What This Work Was
+
+This pass continued the VIOLET plan after `V3.4c` by wiring the launcher-side
+shadow path to the live BLUE factor plane.
+
+The goal stayed read-only. BLUE code, BLUE schemas, and BLUE data structures
+were not modified. The change was entirely on the VIOLET side.
+
+## Scope
+
+This pass added a thin shadow-side live-factor attachment and a focused test
+surface:
+
+- `prod/clean_arch/violet/shadow_live_factors.py`
+- `prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py`
+- `prod/launch_dolphin_violet.py`
+
+It reused the existing V3.4c live BLUE source adapter:
+
+- `prod/clean_arch/violet/live_blue_source.py`
+- `prod/clean_arch/violet/live_factor_source.py`
+- `prod/clean_arch/violet/live_factors.py`
+
+## Why This Was Needed
+
+Before this pass, the Violet shadow launcher still had a base-only decision
+path. That meant the `VioletDecisionEngine` could produce a muted decision, but
+it did not yet receive the full live factor plane from BLUE’s published
+surfaces inside the launcher path.
+
+The missing piece was the launcher-side wiring: source live BLUE factors,
+thread them into `decide(...)`, and keep the journaled breakdown faithful to the
+same factor plane BLUE would have seen at that scan.
+
+## What Was Added
+
+### 1. Shadow live-factor helper
+
+`shadow_live_factors.py` now provides two small helpers:
+
+- `build_shadow_live_source(...)`
+- `shadow_decision_step(...)`
+
+`build_shadow_live_source(...)` assembles the read-only BLUE live factor mirror
+for the shadow path. The helper keeps imports lazy so it can be unit-tested
+without dragging in the full launcher import chain.
+
+`shadow_decision_step(...)` runs one muted shadow decision using the live BLUE
+factor plane, then journals the result when the decision is actuated.
+
+### 2. Launcher wiring
+
+`launch_dolphin_violet.py` now:
+
+- builds the live-factor shadow source when shadow mode is enabled
+- passes the live `SizingFactors` into `VioletDecisionEngine.decide(...)`
+- skips the shadow decision instead of silently falling back to base-only when
+  the live factor plane is missing
+- keeps the existing journal path intact
+
+This preserves the existing muted-shadow architecture while making the shadow
+decision reflect the live BLUE factor plane rather than a reduced fallback.
+
+### 3. Focused tests
+
+`test_violet_launcher_shadow_live_factors.py` covers:
+
+- the live-factor helper contract
+- failure propagation from the client factory
+- the shadow decision step with live factors and journaling
+- the no-live-factor skip path
+
+Because the mount was slow under `pytest`, I verified the helper path directly
+with a small execution script instead of waiting on long file-system waits.
+
+## Exactness Rules Followed
+
+The pass stayed conservative:
+
+- no BLUE edits
+- no schema edits
+- no live fallback to a fake factor plane
+- no execution path changes outside the muted shadow branch
+- no silent loss of the live factor breakdown
+
+## Verification
+
+Direct runtime check:
+
+- the new helper built successfully with injected factories
+- the shadow decision step accepted the live factor plane
+- the decision was journaled with the expected Violet journal table
+
+Observed direct result:
+
+- `ok`
+
+`pytest` on this mount was slow and repeatedly stalled in netfs waits, so I did
+not treat that as a code failure.
+

prod/docs/RECENT_VIOLET_34E_2d37ef0.md

@@ -0,0 +1,83 @@
+# RECENT VIOLET 34E — trade/slot-granularity comparison harness
+
+This pass implemented the next numbered comparison item from the main Violet spec:
+the trade/slot-granularity comparison that was still deferred after the V3.4c
+live-source work and shadow journal wiring.
+
+## What existed before
+
+Before this pass, Violet already had:
+
+- `VioletDecisionEngine` producing shadow decisions
+- `VioletDecisionJournal` persisting actuated decisions to `violet_decisions`
+- live-source adapters for BLUE-published factors
+- full sizing parity through `VioletSizer`
+
+What was still missing was a dedicated comparison unit that could collapse those
+journaled decisions into episode-sized runs and compare them against the terminal
+trade surface at the same granularity.
+
+## What was added
+
+I added a new Violet-only comparator:
+
+- [prod/clean_arch/violet/trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/trade_slot_compare.py)
+- [prod/clean_arch/violet/test_violet_trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_trade_slot_compare.py)
+
+The new module provides:
+
+- `DecisionEpisode`: a collapsed run of contiguous shadow decisions for one
+  asset/side
+- `TradeEpisode`: a collapsed terminal trade record grouped by `trade_id`
+- `EpisodeMatch`: a paired decision/trade episode with timing and row-count gaps
+- `TradeSlotComparison`: the full comparison result
+- `compare_trade_slot_granularity(...)`: the top-level collapse-and-compare API
+
+## How it works
+
+The comparator is deliberately narrow:
+
+- it groups decision rows by `asset` + `side`
+- it splits them into episodes when the scan number gap exceeds the configured
+  `max_scan_gap`
+- it groups trade rows by `trade_id`
+- it ignores malformed rows rather than failing on them
+- it matches episodes by asset/side and preserves unmatched decision/trade rows
+
+This is downstream of the existing shadow journal. It does not reimplement ACB,
+EsoF, OB, or sizing math. It only compares the surfaces that already exist.
+
+## Anomaly handling
+
+The comparator rejects or skips bad inputs instead of letting them pollute the
+episode view:
+
+- missing asset or side
+- missing or invalid scan number
+- malformed or missing trade id
+- non-finite numeric fields
+
+That keeps the harness usable against noisy journal extracts and historical trade
+rows that may contain replay artifacts.
+
+## Tests
+
+The new tests cover:
+
+- episode collapse across contiguous decision rows
+- terminal trade dedupe through `trade_id`
+- scan-gap splitting
+- malformed row handling
+- empty-input behavior
+
+Verification on this pass:
+
+- `PYTHONPATH=/mnt/dolphinng5_predict rtk python -m pytest -q /mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_trade_slot_compare.py`
+- result: `3 passed`
+
+## Why this is the right next step
+
+The main Violet plan explicitly defers trade/slot-granularity comparison until the
+shadow side has a comparable execution surface. That condition is now met well
+enough to build the comparison harness without touching BLUE or the live executor.
+

prod/docs/RECENT_VIOLET_TOUCHED_FILES.md

@@ -0,0 +1,48 @@
+# RECENT VIOLET Touched Files
+
+This inventory covers the recent VIOLET 3.4-series commits on this branch.
+It includes all files touched across the series, grouped by commit, so the
+surface is explicit rather than inferred.
+
+## 722fd9f — VIOLET V3.4b/V3e: journal the full-sizing breakdown
+
+- [prod/clean_arch/violet/shadow_journal.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/shadow_journal.py)
+- [prod/clean_arch/violet/test_violet_shadow_journal.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_shadow_journal.py)
+- [prod/clickhouse/violet/22_violet_decisions.sql](/mnt/dolphinng5_predict/prod/clickhouse/violet/22_violet_decisions.sql)
+
+## a632c59 — VIOLET V3.4b: validate live-factor field paths + HZ sourcing adapter
+
+- [prod/clean_arch/violet/live_factor_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/live_factor_source.py)
+- [prod/clean_arch/violet/test_violet_live_factor_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_live_factor_source.py)
+- [prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md](/mnt/dolphinng5_predict/prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md)
+
+## 1ac3f62 — VIOLET V3.4c: read-only BLUE live source parity
+
+- [prod/clean_arch/violet/live_blue_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/live_blue_source.py)
+- [prod/clean_arch/violet/test_violet_live_blue_source.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_live_blue_source.py)
+
+## 16add44 — DOCS: add RECENT VIOLET 34C detail note
+
+- [prod/docs/RECENT_VIOLET_34C_a632c59.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34C_a632c59.md)
+
+## fb34431 — VIOLET V3.4b: launcher shadow live-factor wiring
+
+- [prod/clean_arch/violet/shadow_live_factors.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/shadow_live_factors.py)
+- [prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_launcher_shadow_live_factors.py)
+- [prod/docs/RECENT_VIOLET_34D_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34D_pending.md)
+- [prod/launch_dolphin_violet.py](/mnt/dolphinng5_predict/prod/launch_dolphin_violet.py)
+
+## 2d37ef0 — VIOLET V3.4c: trade-slot comparison harness
+
+- [prod/clean_arch/violet/test_violet_trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/test_violet_trade_slot_compare.py)
+- [prod/clean_arch/violet/trade_slot_compare.py](/mnt/dolphinng5_predict/prod/clean_arch/violet/trade_slot_compare.py)
+- [prod/docs/RECENT_VIOLET_34E_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34E_pending.md)
+
+## Rename / note follow-ups in the same series
+
+- [prod/docs/RECENT_VIOLET_34D_fb34431.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34D_fb34431.md)
+- [prod/docs/RECENT_VIOLET_34E_2d37ef0.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34E_2d37ef0.md)
+- [prod/docs/RECENT_VIOLET_TOUCHED_FILES.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_TOUCHED_FILES.md)
+- [prod/docs/RECENT_VIOLET_34C_a632c59.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34C_a632c59.md)
+- [prod/docs/RECENT_VIOLET_34D_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34D_pending.md)
+- [prod/docs/RECENT_VIOLET_34E_pending.md](/mnt/dolphinng5_predict/prod/docs/RECENT_VIOLET_34E_pending.md)

prod/docs/VIOLET_V34B_LIVE_FACTOR_FIELD_VALIDATION.md

@@ -0,0 +1,61 @@
+# VIOLET V3.4b — live-factor field-path validation
+
+**Date:** 2026-06-16
+**Task:** validate `prod/clean_arch/violet/live_factors.py`'s candidate field paths
+against how live BLUE actually sources the five sizing multipliers, *before* wiring
+the launcher-sourcing half of V3.4b.
+
+## TL;DR
+
+`live_factors.py` assumes the eight sizing inputs arrive as flat/nested keys in a
+single `hz_snapshot` dict. **That premise holds for only one of them (`posture`).**
+`esof_score` is present in HZ but as a *payload to parse*, not a flat key. The other
+five (`boost`, `beta`, `mc_scale`, `ob_median_imbalance`, `ob_agreement_pct`,
+`dc_status`) are **BLUE-organ outputs that are not published to any HZ map** — they
+live in the live `NDAlphaEngine`'s process memory / are recomputed per scan.
+
+The speculative alternate paths in `live_factors.py` (`acb_boost`, `s_acb_boost`,
+`("acb","boost")`, `day_mc_scale`, `("esof","advisory_score")`, `("ob","market",…)`,
+`("signal","dc_status")`, `safety_posture`, …) **correspond to nothing in live BLUE.**
+They are harmless (first-match-wins, flat canonical key is tried first) but dead.
+
+## Per-factor validated sourcing
+
+Source of truth: `esf_alpha_orchestrator.py` (the composition), `adaptive_circuit_breaker.py`
+(ACB), `nautilus_event_trader.py` (the HZ reads/publishes).
+
+| Factor | How live BLUE gets it | In a VIOLET-readable HZ map? |
+|---|---|---|
+| `posture` | `_day_posture`, set in `begin_day(posture=…)`; published in `engine_snapshot['posture']` (`nautilus_event_trader.py:5097`, map `DOLPHIN_STATE_BLUE`) and `DOLPHIN_SAFETY.latest.posture` | ✅ **flat key `posture`** in engine_snapshot |
+| `esof_score` | `_read_esof_payload()` → `DOLPHIN_FEATURES['esof_latest'\|'esof_advisor_latest']` → `parse_esof_payload` → `esof_score_from_payload(..., max_age_s=ESOF_FRESHNESS_S)` (`nautilus_event_trader.py:707-719`) | ✅ but a **payload parse**, not a flat `esof_score` |
+| `boost` / `beta` | `acb.get_dynamic_boost_from_hz(date)` → `acb_info['boost'\|'beta']`; the ACB **computes** them from `DOLPHIN_FEATURES['exf_latest']` (funding_btc/dvol_btc/fng/taker — `adaptive_circuit_breaker.py:511,528-533`). Applied via `begin_day` / `update_acb_boost` (`esf_alpha_orchestrator.py:764,946`). | ❌ raw inputs are in HZ; the **scalar requires running the ACB** |
+| `mc_scale` | `_day_mc_scale`, **derived** in `begin_day` from MC-Forewarner `mc_orange`/`mc_red` flags (`esf_alpha_orchestrator.py:962-964`: orange→0.5, red/TURTLE/HIBERNATE→…) | ❌ not a HZ scalar |
+| `ob_median_imbalance` / `ob_agreement_pct` | `ob_engine.get_market(bar_idx, symbols)` over the live OB feed, **per asset** (`esf_alpha_orchestrator.py:590-595`) | ❌ computed live; not in HZ |
+| `dc_status` | per-asset `signal.dc_status` from the signal generator (`esf_alpha_orchestrator.py:576`) | ❌ computed per-scan; not in HZ |
+
+`engine_snapshot` payload (the map BLUE publishes for consumers) was inspected in full
+(`nautilus_event_trader.py:5092-5126`): it carries `posture`, `last_vel_div`, `vol_ok`,
+`last_scan_number`, `capital`, leverage caps, position list — **and none of the five
+organ-derived multipliers.**
+
+## Consequence for V3.4b sourcing (#2)
+
+A faithful, *complete* live-factor source is NOT a HZ scrape — it requires VIOLET to run
+the same organs BLUE does:
+- an **ACB** over `DOLPHIN_FEATURES['exf_latest']` → boost/beta,
+- the **MC** flag→`mc_scale` derivation,
+- an **OBFeatureEngine** over the OB feed → ob_*,
+- a **signal generator** → dc_status.
+
+That is a multi-organ sprint (call it **V3.4c**), not a quick wiring.
+
+What IS sourceable now, from maps BLUE already publishes, read-only:
+- **`posture`** ← `engine_snapshot['posture']`
+- **`esof_score`** ← `DOLPHIN_FEATURES['esof_latest']` via BLUE's own `esof_score_from_payload`
+
+So the honest V3.4b increment (this PR) is a **pure adapter** —
+`live_factor_source.py` — that sources those two faithfully and supplies BLUE's own
+neutral sentinels for the organ-derived five (`boost=1.0`, `beta=0.0`, `mc_scale=1.0`,
+`ob_*=None`, `dc_status="NONE"`), feeding `extract_live_sizing_factors`. The shadow
+journal's V3.4 breakdown then records posture+esof live and the rest neutral — explicit,
+not silently faked. The organ wiring (boost/beta/mc/ob/dc live) is deferred to V3.4c.

prod/launch_dolphin_violet.py

@@ -50,6 +50,10 @@ from prod.launch_dolphin_pink import (  # noqa: E402
     _resolve_bingx_exchange_leverage_cap,
     _resolve_bingx_recv_window_ms,
 )
+from prod.clean_arch.violet.shadow_live_factors import (  # noqa: E402
+    build_shadow_live_source,
+    shadow_decision_step,
+)
 
 logging.basicConfig(
     level=logging.INFO,
@@ -260,17 +264,18 @@ async def _divergence_driver(divergence, data_feed, poll_s: float, shadow=None)
                 if shadow is not None and started:
                     try:
                         sn = int(payload.get("scan_number") or 0)
-                        shadow["engine"].observe(payload, sn)
                         vd = payload.get("vel_div")
                         if vd is not None:
                             now_ns = shadow["mono_ns"]()
-                            d = shadow["engine"].decide(
-                                now_ns=now_ns, scan_number=sn,
-                                capital=shadow["capital"], vel_div=float(vd),
+                            if shadow_decision_step(
+                                shadow,
+                                payload,
+                                scan_number=sn,
+                                now_ns=now_ns,
+                                vel_div=float(vd),
                                 vol_ok=bool(payload.get("vol_ok", True)),
-                            )
-                            if d is not None:
-                                shadow["journal"].journal(d, mono_ns=now_ns)
+                            ):
+                                shadow["live_decisions"] += 1
                     except Exception as exc:  # noqa: BLE001 — shadow must never die
                         LOGGER.debug("shadow decision failed: %s", exc)
         except Exception as exc:  # noqa: BLE001 — sampling must never die
@@ -333,13 +338,29 @@ def _build_shadow():
     relaxed = abs(thr - (-0.02)) > 1e-9
     engine = VioletDecisionEngine(entry_vel_div_threshold=thr)
     journal = VioletDecisionJournal(sink=ch_put_violet, session_id=sess)
+    try:
+        live_source = build_shadow_live_source()
+    except Exception as exc:
+        LOGGER.warning(
+            "VIOLET shadow live-factor source unavailable (%s) — shadow DISABLED.",
+            exc,
+        )
+        return None
     LOGGER.warning(
         "VIOLET DECISION SHADOW ON (session=%s ref_capital=%.0f entry_thr=%.4f%s) — "
         "journaling muted decisions to dolphin_violet.violet_decisions; NO orders.",
         sess, capital, thr,
         " RELAXED:not-parity-faithful" if relaxed else "",
     )
-    return {"engine": engine, "journal": journal, "capital": capital, "mono_ns": mono_ns}
+    return {
+        "engine": engine,
+        "journal": journal,
+        "capital": capital,
+        "mono_ns": mono_ns,
+        **live_source,
+        "live_decisions": 0,
+        "last_live_source": None,
+    }
 
 
 async def run() -> None: