diff --git a/prod/bingx/journal.py b/prod/bingx/journal.py
index 32686ce..7a7c0b6 100644
--- a/prod/bingx/journal.py
+++ b/prod/bingx/journal.py
@@ -12,6 +12,7 @@ from typing import Any
from prod.ch_writer import ch_put
from prod.ch_writer import ch_put_green
from prod.ch_writer import ch_put_prodgreen
+from prod.ch_writer import ch_put_violet
from prod.ch_writer import ch_put_pink
# ─── Account event rate control (§10.2) ──────────────────────────────────────
@@ -175,6 +176,7 @@ _STRATEGY_DB_MAP: dict[str, str] = {
"green": "dolphin_green",
"prodgreen": "dolphin_prodgreen",
"pink": "dolphin_pink",
+ "violet": "dolphin_violet",
}
_STRATEGY_SINK_MAP: dict[str, Any] = {
@@ -182,6 +184,7 @@ _STRATEGY_SINK_MAP: dict[str, Any] = {
"green": ch_put_green,
"prodgreen": ch_put_prodgreen,
"pink": ch_put_pink,
+ "violet": ch_put_violet,
}
_STRATEGY_SINK_NAME_MAP: dict[str, str] = {
@@ -189,6 +192,7 @@ _STRATEGY_SINK_NAME_MAP: dict[str, str] = {
"green": "ch_put_green",
"prodgreen": "ch_put_prodgreen",
"pink": "ch_put_pink",
+ "violet": "ch_put_violet",
}
diff --git a/prod/ch_writer.py b/prod/ch_writer.py
index d02493c..541c150 100644
--- a/prod/ch_writer.py
+++ b/prod/ch_writer.py
@@ -547,6 +547,7 @@ _writer = _CHWriter(db="dolphin")
_writer_green = _CHWriter(db="dolphin_green")
_writer_prodgreen = _CHWriter(db="dolphin_prodgreen")
_writer_pink = _CHWriter(db="dolphin_pink")
+_writer_violet = _CHWriter(db="dolphin_violet")
def ch_put(table: str, row: dict) -> None:
@@ -577,6 +578,15 @@ def ch_put_pink(table: str, row: dict) -> None:
_writer_pink.put(table, row)
+def ch_put_violet(table: str, row: dict) -> None:
+ """
+ Fire-and-forget insert into dolphin_violet.
(VIOLET / observe-only
+ rebuild runtime). Namespace isolation is a stage gate: VIOLET must never
+ write to dolphin or dolphin_pink.
+ """
+ _writer_violet.put(table, row)
+
+
# ─── V7 decision journal (PINK §10 data volume / §37 routing) ────────────────
PINK_V7_JOURNAL_DB = "dolphin_pink"
V7_DECISION_TABLE = "v7_decision_events"
diff --git a/prod/clean_arch/violet/test_violet_namespace_isolation.py b/prod/clean_arch/violet/test_violet_namespace_isolation.py
new file mode 100644
index 0000000..8b2f962
--- /dev/null
+++ b/prod/clean_arch/violet/test_violet_namespace_isolation.py
@@ -0,0 +1,124 @@
+"""V1 GATE: VIOLET namespace isolation.
+
+Zero VIOLET writes may reach dolphin (BLUE) or dolphin_pink databases, BLUE
+or PINK Hazelcast maps, or PINK Zinc regions. This suite pins:
+ (a) the violet CH sink targets dolphin_violet exclusively;
+ (b) prod.bingx.journal resolves "violet" explicitly and does NOT hit the
+ legacy fallback that routes unknown strategies into BLUE's dolphin DB;
+ (c) Zinc region names for prefix "violet" are disjoint from "pink";
+ (d) the persistence layer wired with the violet sink writes only to
+ dolphin_violet.
+"""
+
+from __future__ import annotations
+
+import sys
+
+sys.path.insert(0, "/mnt/dolphinng5_predict")
+sys.path.insert(0, "/mnt/dolphinng5_predict/nautilus_dolphin")
+
+import pytest
+
+
+def test_ch_put_violet_targets_dolphin_violet(monkeypatch):
+ import prod.ch_writer as chw
+
+ captured = []
+ monkeypatch.setattr(
+ chw._writer_violet, "put", lambda table, row: captured.append(("dolphin_violet", table))
+ )
+ monkeypatch.setattr(
+ chw._writer, "put", lambda table, row: captured.append(("dolphin", table))
+ )
+ monkeypatch.setattr(
+ chw._writer_pink, "put", lambda table, row: captured.append(("dolphin_pink", table))
+ )
+ chw.ch_put_violet("status_snapshots", {"x": 1})
+ assert captured == [("dolphin_violet", "status_snapshots")]
+
+
+def test_journal_sink_resolves_violet_not_blue_fallback():
+ """The line-~210 fallback returns ch_put (BLUE dolphin DB) for unknown
+ strategies — 'violet' must resolve explicitly, never via fallback."""
+ from prod.bingx import journal as j
+ from prod.ch_writer import ch_put, ch_put_violet
+
+ assert j._STRATEGY_DB_MAP.get("violet") == "dolphin_violet"
+ sink = j._sink_for_strategy("violet")
+ assert sink is ch_put_violet
+ assert sink is not ch_put
+ assert j._db_for_strategy("violet") == "dolphin_violet"
+ # regression: a truly-unknown strategy still falls back to BLUE — the
+ # hazard the violet entry exists to avoid (documents the footgun).
+ assert j._sink_for_strategy("chartreuse") is ch_put
+
+
+def test_zinc_region_names_disjoint_from_pink():
+ from prod.clean_arch.dita_v2.real_zinc_plane import RealZincPlane
+
+ # Region names derive from the prefix; compute without creating regions.
+ def names(prefix: str):
+ base = prefix.strip("/").replace("/", "_")
+ return {f"{base}_intent", f"{base}_state", f"{base}_control"}
+
+ assert names("violet").isdisjoint(names("pink"))
+ # the derivation rule above must match the class's own (guard against
+ # silent renames in real_zinc_plane)
+ import inspect
+
+ src = inspect.getsource(RealZincPlane.__init__)
+ for piece in ("_intent", "_state", "_control"):
+ assert piece in src
+
+
+def test_persistence_with_violet_sink_writes_only_dolphin_violet(monkeypatch):
+ """PinkClickHousePersistence(sink=ch_put_violet) — every row lands in
+ dolphin_violet; zero rows reach the BLUE/PINK writers."""
+ import prod.ch_writer as chw
+ from prod.clean_arch.persistence.pink_clickhouse import PinkClickHousePersistence
+
+ dbs = []
+ monkeypatch.setattr(chw._writer_violet, "put", lambda t, r: dbs.append("dolphin_violet"))
+ monkeypatch.setattr(chw._writer, "put", lambda t, r: dbs.append("dolphin"))
+ monkeypatch.setattr(chw._writer_pink, "put", lambda t, r: dbs.append("dolphin_pink"))
+ monkeypatch.setattr(chw._writer_green, "put", lambda t, r: dbs.append("dolphin_green"))
+ monkeypatch.setattr(chw._writer_prodgreen, "put", lambda t, r: dbs.append("dolphin_prodgreen"))
+
+ from prod.clean_arch.dita_v2.account import AccountProjection
+
+ p = PinkClickHousePersistence(
+ AccountProjection(), sink=chw.ch_put_violet, v7_sink=chw.ch_put_violet
+ )
+ # Drive a representative write path: the recovery/state writer.
+ try:
+ p.persist_recovery_state(snapshot=None, acc_dict={
+ "capital": 25_000.0, "equity": 25_000.0, "open_positions": 0,
+ })
+ except Exception:
+ pass # row-shape errors are fine — we only assert ROUTING
+ assert "dolphin" not in dbs
+ assert "dolphin_pink" not in dbs
+ assert set(dbs) <= {"dolphin_violet"}
+
+
+def test_violet_hz_map_names():
+ """The V1 launcher's defaults must be violet-namespaced (string contract
+ pinned here before the launcher lands in C4)."""
+ expected = {
+ "state_map": "DOLPHIN_STATE_VIOLET",
+ "pnl_map": "DOLPHIN_PNL_VIOLET",
+ "journal_db": "dolphin_violet",
+ "journal_strategy": "violet",
+ "trader_id": "DOLPHIN-VIOLET-001",
+ "strategy_name": "violet",
+ }
+ try:
+ from prod.launch_dolphin_violet import VIOLET_DEFAULTS
+ except ImportError:
+ pytest.skip("launcher lands in C4 — contract pinned for then")
+ for k, v in expected.items():
+ assert VIOLET_DEFAULTS[k] == v
+
+
+if __name__ == "__main__":
+ raise SystemExit(pytest.main([__file__, "-v"]))