dita_v2: reconcile SOA — vendored thread-safety hardening ∪ venue plane

3-way merge (base=pre-venue-plane upstream, ours=violet-main vendored hardening,
theirs=upstream venue plane): RLock atomic-snapshot wrapping (account.py,
real_control_plane, real_zinc_plane), lazy __getattr__ imports (__init__),
account-core test coverage — merged with venue_region telemetry. Zero conflicts;
all files AST-verified. Ends the two-way vendor drift found in
DITAV2_SOA_SURVEY_20260702 §5 / UV_DITAV2_SOA_VERDICT_20260703.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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,6 +60,7 @@ class AccountProjection:
         GIL guarantees single-field reference assignment is atomic, so readers
         that hold snap = kernel.account.snapshot before use see a consistent view.
         """
+        with self._lock:
             cur = self.snapshot
             self.snapshot = AccountSnapshot(
                 capital=kw.get("capital", cur.capital),
@@ -75,7 +77,11 @@ class AccountProjection:
                 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:
+        with self._lock:
             open_positions = 0
             open_notional = 0.0
             unrealized_pnl = 0.0
@@ -88,12 +94,14 @@ class AccountProjection:
                     mark = safe_float(slot.metadata.get("mark_price"), mark)
                     open_notional += abs(slot.size) * abs(mark)
                     unrealized_pnl += float(slot.unrealized_pnl or 0.0)
+            capital = self.snapshot.capital
+            peak_capital = self.snapshot.peak_capital
             self._replace_snapshot(
                 open_positions=open_positions,
                 open_notional=open_notional,
                 unrealized_pnl=unrealized_pnl,
-            equity=self.snapshot.capital + unrealized_pnl if math.isfinite(self.snapshot.capital + unrealized_pnl) else self.snapshot.capital,
+                equity=capital + unrealized_pnl if math.isfinite(capital + unrealized_pnl) else capital,
-            peak_capital=max(self.snapshot.peak_capital, self.snapshot.capital) if open_notional > 0 and self.snapshot.capital > 0 else self.snapshot.peak_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:
@@ -106,6 +114,7 @@ class AccountProjection:
         Guards: wallet_balance must be > 0 and finite (the zero-wb frame lesson
         from ACCOUNT_UPDATE frames with no USDT balance entry).
         """
+        with self._lock:
             wb = safe_float(wallet_balance, 0.0)
             if wb <= 0.0 or not math.isfinite(wb):
                 return
@@ -119,24 +128,26 @@ class AccountProjection:
             self.snapshot.peak_capital = max(self.snapshot.peak_capital, wb)
 
     def settle(self, realized_pnl: float, fees: float = 0.0) -> None:
+        with self._lock:
+            cur = self.snapshot
             rp = safe_float(realized_pnl, 0.0)
             # Include fees in capital delta (today fees only accumulate in
             # fees_paid while published capital ignores them between reseeds).
             net = rp - safe_float(fees, 0.0)
-        new_capital = safe_float(self.snapshot.capital + net, self.snapshot.capital)
+            new_capital = safe_float(cur.capital + net, cur.capital)
             if self.max_capital is not None:
                 new_capital = min(new_capital, self.max_capital)
             new_capital = max(self.min_capital, new_capital)
-        new_source = self.snapshot.capital_source
+            new_source = cur.capital_source
             if new_source == "e_anchored" and abs(net) > 1e-12:
                 new_source = "k_bridged"
-        new_fees = self.snapshot.fees_paid + safe_float(fees, 0.0)
+            new_fees = cur.fees_paid + safe_float(fees, 0.0)
-        new_equity = new_capital + self.snapshot.unrealized_pnl
+            new_equity = new_capital + cur.unrealized_pnl
             if not math.isfinite(new_equity):
                 new_equity = new_capital
             self._replace_snapshot(
                 capital=new_capital, capital_source=new_source,
-            realized_pnl=self.snapshot.realized_pnl + rp,
+                realized_pnl=cur.realized_pnl + rp,
                 fees_paid=new_fees, equity=new_equity,
             )
 
@@ -154,6 +165,7 @@ class AccountProjection:
         bars_held: int = 0,
         metadata: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
+        with self._lock:
             self.snapshot.equity = self.snapshot.capital + self.snapshot.unrealized_pnl
             return {
                 "timestamp": timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp),
@@ -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,6 +358,7 @@ class AccountProjectionV2:
         fee: float,
         realized_pnl: float,
     ) -> None:
+        with self._lock:
             self._k_realized += _safe(realized_pnl)
             self._k_fees += _safe(fee)
             self._e_last_fill_price = _safe(fill_price)
@@ -353,6 +367,7 @@ class AccountProjectionV2:
             self._e_last_fill_realized = _safe(realized_pnl)
 
     def apply_funding(self, amount: float) -> None:
+        with self._lock:
             self._k_funding += _safe(amount)
             self._e_last_funding = _safe(amount)
 
@@ -364,12 +379,14 @@ class AccountProjectionV2:
         used_margin: float,
         maint_margin: float,
     ) -> None:
+        with self._lock:
             self._e_wallet_balance = _safe(wallet_balance)
             self._e_avail_margin = _safe(available_margin)
             self._e_used_margin = _safe(used_margin)
             self._e_maint_margin = _safe(maint_margin)
 
     def apply_position_update(self, positions: List[EPosition]) -> None:
+        with self._lock:
             self._e_positions = list(positions)
 
     # ------------------------------------------------------------------
@@ -382,6 +399,7 @@ class AccountProjectionV2:
         slots: Iterable[TradeSlot],
         ts: Optional[float] = None,
     ) -> AccountSnapshotV2:
+        with self._lock:
             self._event_seq += 1
             snap = self._build(self._event_seq, source_event_id, list(slots), ts or time.time())
             self._snapshot = snap
@@ -389,10 +407,12 @@ class AccountProjectionV2:
 
     @property
     def snapshot(self) -> AccountSnapshotV2:
+        with self._lock:
             return self._snapshot
 
     @property
     def k_capital(self) -> float:
+        with self._lock:
             raw = self._seed + self._k_realized - self._k_fees - self._k_funding
             if self._max_capital is not None:
                 raw = min(raw, self._max_capital)

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,6 +89,7 @@ class RealZincControlPlane(ControlPlane):
         self.region.close()
 
     def read(self) -> KernelControlSnapshot:
+        with self._lock:
             payload = _decode_packet(self.region.as_buffer())
             control = payload.get("control") if isinstance(payload, dict) else None
             if not isinstance(control, dict):
@@ -94,12 +98,14 @@ class RealZincControlPlane(ControlPlane):
             return self._snapshot
 
     def update(self, update: ControlUpdate) -> KernelControlSnapshot:
+        with self._lock:
             self._snapshot = update.apply(self.read())
             self._seq += 1
             self._write_region(self._seq, self._snapshot.as_dict())
             return self._snapshot
 
     def mirror(self) -> Dict[str, Any]:
+        with self._lock:
             return self._snapshot.as_dict()
 
     def wait(self, timeout_ms: int = 1000) -> bool:

prod/clean_arch/dita_v2/real_zinc_plane.py

@@ -258,11 +258,13 @@ class RealZincPlane:
             self._write_region(self.state_region, self._state_seq, payload)
 
     def read_slots(self) -> List[TradeSlot]:
+        with self._lock:
             payload = _decode_packet(self.state_region.as_buffer())
             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]]:
+        with self._lock:
             payload = _decode_packet(self.intent_region.as_buffer())
             items = payload.get("items", []) if isinstance(payload, dict) else []
             return list(items)
@@ -274,6 +276,7 @@ class RealZincPlane:
             self._write_region(self.control_region, self._control_seq, {"control": control.as_dict()})
 
     def read_control(self) -> KernelControlSnapshot:
+        with self._lock:
             payload = _decode_packet(self.control_region.as_buffer())
             control = payload.get("control") if isinstance(payload, dict) else None
             if not isinstance(control, dict):

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: