dita_v2(test): repair test_double_buffer control test — suite was un-collectable

Line 154 held mutant debris: KernelControlSnapshot(mode="live", mode="live"") — duplicate
kwarg + unterminated string. Committed since a55501b, it made the ENTIRE dita_v2 suite fail
collection, so nobody has run it. That is why no test caught the bingx_venue telemetry
signature skew that orphaned 6 live positions tonight.

The test was fantasy besides: asserted read.arming == "DARK" (no such field) and
mode == "live" (KernelMode is NORMAL|DEBUG). Rewritten to assert a real invariant —
the second write must flip the buffer and become visible.

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

prod/clean_arch/dita_v2/test_double_buffer.py

@@ -0,0 +1,231 @@
+"""Phase-4 tests: double-buffer region — torn-read impossible under concurrent publish.
+
+Key invariants:
+1. read() never returns a partially-updated value
+2. After write(), subsequent reads return the written value
+3. Concurrent hammering: reads under heavy write load never get torn data
+4. Deep-copy isolation: read returns independent copy
+"""
+
+from __future__ import annotations
+
+import sys
+import threading
+import time
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+
+from prod.clean_arch.dita_v2.contracts import AccountStateSnapshot, VenueTelemetrySnapshot
+from prod.clean_arch.dita_v2.double_buffer import DoubleBufferRegion
+from prod.clean_arch.dita_v2.zinc_plane import InMemoryZincPlane
+
+
+class TestDoubleBufferBasic:
+    """Basic write/read cycle with DoubleBufferRegion."""
+
+    def test_write_then_read(self) -> None:
+        buf = DoubleBufferRegion(lambda: AccountStateSnapshot())
+        snap = AccountStateSnapshot(wallet_balance=25000.0, e_live=True)
+        buf.write(snap)
+        read = buf.read()
+        assert read.wallet_balance == 25000.0
+        assert read.e_live is True
+
+    def test_overwrite(self) -> None:
+        buf = DoubleBufferRegion(lambda: AccountStateSnapshot())
+        buf.write(AccountStateSnapshot(wallet_balance=100.0, e_live=True))
+        assert buf.read().wallet_balance == 100.0
+        buf.write(AccountStateSnapshot(wallet_balance=200.0, e_live=True))
+        assert buf.read().wallet_balance == 200.0
+
+    def test_initial_read_is_default(self) -> None:
+        buf = DoubleBufferRegion(lambda: AccountStateSnapshot(e_live=False))
+        assert buf.read().e_live is False
+        assert buf.read().wallet_balance == 0.0
+
+    def test_read_returns_deep_copy(self) -> None:
+        """read() returns a copy — modifying it doesn't corrupt the buffer."""
+        buf = DoubleBufferRegion(lambda: AccountStateSnapshot(wallet_balance=100.0))
+        r1 = buf.read()
+        # Attempt to mutate the returned object (frozen, so conceptually)
+        assert r1.wallet_balance == 100.0
+        buf.write(AccountStateSnapshot(wallet_balance=200.0))
+        # Previous read result unaffected
+        assert r1.wallet_balance == 100.0
+
+
+class TestDoubleBufferTornRead:
+    """Under heavy concurrent publish/read, no torn reads.
+
+    Mutation: remove double-buffer (use single slot) → reads may see partial
+    state under write load.
+    """
+
+    def test_no_torn_read_under_writer(self) -> None:
+        """1000 rapid writes interleaved with 1000 reads — all reads return
+        a consistent snapshot that was actually published."""
+        plane = InMemoryZincPlane()
+        seen: list[AccountStateSnapshot] = []
+        errors: list[str] = []
+        lock = threading.Lock()
+
+        def writer() -> None:
+            for i in range(100):
+                plane.publish_account(AccountStateSnapshot(
+                    wallet_balance=float(i * 100),
+                    available_margin=float(i * 80),
+                    used_margin=float(i * 20),
+                    event_seq=i, mono_ns=i,
+                    e_live=True, reconcile_ok=(i % 2 == 0),
+                ))
+
+        def reader() -> None:
+            for _ in range(100):
+                try:
+                    snap = plane.read_account()
+                    # A torn read would show inconsistent fields.
+                    # wallet_balance should always be a multiple of 100
+                    # between 0 and 9900
+                    wb = snap.wallet_balance
+                    if isinstance(wb, float):
+                        pass
+                except Exception as e:
+                    with lock:
+                        errors.append(str(e))
+
+        threads = [threading.Thread(target=writer, daemon=True) for _ in range(5)]
+        threads += [threading.Thread(target=reader, daemon=True) for _ in range(10)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=10)
+        assert len(errors) == 0, f"errors: {errors}"
+
+    def test_no_torn_read_rapid_alternating(self) -> None:
+        """Interleaved write/read at high frequency — never returns garbage."""
+        plane = InMemoryZincPlane()
+        for i in range(5000):
+            plane.publish_account(AccountStateSnapshot(
+                wallet_balance=float(i * 1000),
+                available_margin=float(i * 800),
+                used_margin=float(i * 200),
+                event_seq=i, mono_ns=i,
+                e_live=True, reconcile_ok=True,
+            ))
+            snap = plane.read_account()
+            # Must be internally consistent: all fields match same event_seq
+            assert snap.event_seq == i, f"seq mismatch: {snap.event_seq} != {i}"
+            assert snap.wallet_balance == float(i * 1000), f"wb mismatch"
+        assert plane.read_account().event_seq == 4999
+
+    def test_concurrent_no_torn_on_venue(self) -> None:
+        """Venue region double-buffer also protects against torn reads."""
+        plane = InMemoryZincPlane()
+        errors: list[str] = []
+        lock = threading.Lock()
+
+        def vwriter() -> None:
+            for i in range(100):
+                plane.publish_venue(VenueTelemetrySnapshot(
+                    phase="live", status=f"S{i}", venue="bingx",
+                ))
+
+        def vreader() -> None:
+            for _ in range(100):
+                try:
+                    snap = plane.read_venue()
+                    _ = snap.phase
+                except Exception as e:
+                    with lock:
+                        errors.append(str(e))
+
+        threads = [threading.Thread(target=vwriter, daemon=True) for _ in range(5)]
+        threads += [threading.Thread(target=vreader, daemon=True) for _ in range(5)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=10)
+        assert len(errors) == 0
+
+    def test_control_region_double_buffered(self) -> None:
+        """Control region round-trips, and a second write flips the buffer."""
+        from prod.clean_arch.dita_v2.control import KernelControlSnapshot, KernelMode
+
+        plane = InMemoryZincPlane()
+        plane.update_control(KernelControlSnapshot(mode=KernelMode.DEBUG))
+        assert plane.read_control().mode is KernelMode.DEBUG
+        # The buffer must FLIP: a later write has to become visible, not be
+        # shadowed by the first snapshot.
+        plane.update_control(KernelControlSnapshot(mode=KernelMode.NORMAL))
+        assert plane.read_control().mode is KernelMode.NORMAL
+
+    def test_two_regions_independent(self) -> None:
+        """Account and venue regions don't interfere."""
+        plane = InMemoryZincPlane()
+        plane.publish_account(AccountStateSnapshot(wallet_balance=100.0, e_live=True))
+        plane.publish_venue(VenueTelemetrySnapshot(phase="test"))
+        assert plane.read_account().wallet_balance == 100.0
+        assert plane.read_venue().phase == "test"
+
+    def test_deep_copy_independence(self) -> None:
+        """Multiple reads from the same buffer return independent snapshots."""
+        from prod.clean_arch.dita_v2.contracts import VenueTelemetrySnapshot
+        buf = DoubleBufferRegion(lambda: VenueTelemetrySnapshot())
+        buf.write(VenueTelemetrySnapshot(phase="original"))
+        r1 = buf.read()
+        r2 = buf.read()
+        assert r1.phase == "original"
+        assert r2.phase == "original"
+        # They should be equal but independent
+        assert r1 == r2
+
+
+class TestDoubleBufferConcurrent:
+    """Heavy concurrent stress on DoubleBufferRegion."""
+
+    def test_50_writers_50_readers(self) -> None:
+        """50 threads writing, 50 threads reading — no crash."""
+        from prod.clean_arch.dita_v2.contracts import VenueTelemetrySnapshot
+        buf = DoubleBufferRegion(lambda: VenueTelemetrySnapshot())
+        errors: list[Exception] = []
+        lock = threading.Lock()
+
+        def writer() -> None:
+            try:
+                for i in range(50):
+                    buf.write(VenueTelemetrySnapshot(phase=f"run_{i}"))
+            except Exception as e:
+                with lock:
+                    errors.append(e)
+
+        def reader() -> None:
+            try:
+                for _ in range(50):
+                    _ = buf.read()
+            except Exception as e:
+                with lock:
+                    errors.append(e)
+
+        threads = [threading.Thread(target=writer, daemon=True) for _ in range(50)]
+        threads += [threading.Thread(target=reader, daemon=True) for _ in range(50)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=15)
+        assert len(errors) == 0, f"{len(errors)} errors"
+
+    def test_rapid_alternating_10k(self) -> None:
+        """10k alternating write/read on a fresh buffer — always returns a valid value."""
+        buf = DoubleBufferRegion(lambda: AccountStateSnapshot())
+        for i in range(5000):
+            buf.write(AccountStateSnapshot(
+                wallet_balance=float(i * 100),
+                available_margin=float(i * 80),
+                used_margin=float(i * 20),
+                event_seq=i, mono_ns=i,
+                e_live=True, reconcile_ok=(i % 2 == 0),
+            ))
+            snap = buf.read()
+            assert snap.event_seq == i
+            assert snap.wallet_balance == float(i * 100)
+        assert buf.read().event_seq == 4999