exec(unified): L2 foundation — contract §2 + whether-maker Router §6, pure, 26 tests

First code of the Unified Execution Layer (SPEC_UNIFIED_EXEC_LAYER_20260714.md). Pure
policy, stdlib+Decimal only, zero I/O, zero venue knowledge, zero importers elsewhere —
adopting it breaks nothing (freeze-safe; operator unparked the build 2026-07-14).

- contract.py: ExecutionRequest input surface (§2.1) + V-TYPES (Side, UrgencyClass,
  ProtectiveSpec, ExecutionAdvice), frozen, validated-at-construction, illegal states
  unrepresentable. 'an asset, a size, and a prayer'.
- router.py: decide(request) -> RoutingDecision — total pure map of the §6 urgency
  ladder. Encodes A1 adjudication verdict in the architecture: Router=whether-maker,
  SmartPlacer=where-in-book via the wants_placement/pre_submit seam. Complementary.
- _constants.py: policy magnitudes with provenance; PROVISIONAL ones flagged for
  L8/L10 calibration (never vibes, never a hardcoded cadence).
- test_exec_unified.py: 26 behaviour + mutation-litmus tests. Verified RED under
  CATASTROPHIC->MAKER and ACQUIRE cross_on_expiry->True mutations.

Not yet wired: PINK drive-loop port (§7), venue dialect (§11), telemetry (§12).

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

prod/exec_unified/test_exec_unified.py

@@ -0,0 +1,200 @@
+"""Behaviour + mutation-litmus tests for the Unified Execution Layer core.
+
+Testing doctrine (prod/docs/TESTING_DOCTRINE.md): these assert the VALUE/SHAPE/INVARIANT,
+not "it ran". Each router test is written so that breaking the implementation goes RED —
+the mutation litmus is spelled out per test. Run:
+
+    /home/dolphin/siloqy_env/bin/python3 -m pytest prod/exec_unified/test_exec_unified.py -q
+
+Mutation examples that MUST turn something red:
+  * CATASTROPHIC method MAKER→...            -> test_catastrophic_crosses_now
+  * ACQUIRE cross_on_expiry False→True       -> test_acquire_abandons_never_crosses
+  * PROTECT max_reprices 1→2                  -> test_protect_one_reprice_then_cross
+  * drop a urgency branch in decide()         -> test_decide_total_over_all_urgencies
+  * ignore_advice True on a non-CATASTROPHIC  -> test_only_catastrophic_ignores_advice
+"""
+from __future__ import annotations
+
+from decimal import Decimal
+
+import pytest
+
+from prod.exec_unified import (
+    ExecutionAdvice,
+    ExecutionMethod,
+    ExecutionRequest,
+    ProtectiveSpec,
+    RoutingDecision,
+    Side,
+    TtlDiscipline,
+    UrgencyClass,
+    decide,
+)
+from prod.exec_unified import _constants as K
+
+
+def _req(urgency: UrgencyClass, **kw) -> ExecutionRequest:
+    base = dict(
+        request_id="r-1",
+        asset="BTCUSDT",
+        side=Side.BUY,
+        size=Decimal("0.001"),
+        urgency=urgency,
+    )
+    # ACQUIRE is the only non-reduce_only default that matters; keep entries clean.
+    if urgency is not UrgencyClass.ACQUIRE:
+        base["reduce_only"] = True
+    base.update(kw)
+    return ExecutionRequest(**base)
+
+
+# ---------------------------------------------------------------- router: the ladder
+
+def test_catastrophic_crosses_now():
+    d = decide(_req(UrgencyClass.CATASTROPHIC))
+    assert d.method is ExecutionMethod.TAKER          # never maker — this is the SL/kill path
+    assert d.max_reprices == 0                         # no chase, no resting
+    assert d.ttl is TtlDiscipline.IMMEDIATE
+    assert d.ignore_advice is True                     # advice inadmissible by law
+    assert d.wants_placement is False                  # no book games on a catastrophe
+
+
+def test_protect_one_reprice_then_cross():
+    d = decide(_req(UrgencyClass.PROTECT))
+    assert d.method is ExecutionMethod.MAKER
+    assert d.max_reprices == 1                          # EXACTLY one (spec §6)
+    assert d.ttl is TtlDiscipline.BOUNDED_MS
+    assert d.max_ms == K.PROTECT_MAX_MS == 2_000        # ≤2 s total
+    assert d.cross_on_expiry is True                    # then cross — a protect must complete
+
+
+def test_harvest_chases_then_gives_up_by_crossing():
+    d = decide(_req(UrgencyClass.HARVEST))
+    assert d.method is ExecutionMethod.MAKER
+    assert d.max_reprices == K.HARVEST_MAX_CHASES
+    assert d.wants_placement is True                    # SmartPlacer sets the target±offset
+    assert d.giveup_regression_frac == K.HARVEST_GIVEUP_REGRESSION_FRAC
+    assert d.cross_on_expiry is True                    # give-up still wants the harvest
+
+
+def test_rotate_uses_caller_deadline():
+    d = decide(_req(UrgencyClass.ROTATE, deadline_ms=60_000))
+    assert d.method is ExecutionMethod.MAKER
+    assert d.ttl is TtlDiscipline.DEADLINE
+    assert d.cross_on_expiry is True
+
+
+def test_acquire_abandons_never_crosses():
+    # THE distinctive invariant: a missed entry is free, a chased/crossed entry is not (§4-1).
+    d = decide(_req(UrgencyClass.ACQUIRE))
+    assert d.method is ExecutionMethod.MAKER
+    assert d.max_reprices == 0                           # never chase
+    assert d.cross_on_expiry is False                    # never cross — abandon
+    assert d.ttl is TtlDiscipline.UNBOUNDED
+
+
+def test_only_catastrophic_ignores_advice():
+    for u in UrgencyClass:
+        d = decide(_req(u, deadline_ms=1000))
+        assert d.ignore_advice is (u is UrgencyClass.CATASTROPHIC), u
+
+
+def test_decide_total_over_all_urgencies():
+    # No urgency may fall through unhandled (drop a branch -> this goes red).
+    for u in UrgencyClass:
+        d = decide(_req(u, deadline_ms=1000))
+        assert isinstance(d, RoutingDecision)
+        assert d.rationale  # every path explains itself
+
+
+def test_decide_is_pure_and_deterministic():
+    r = _req(UrgencyClass.HARVEST)
+    a, b = decide(r), decide(r)
+    assert a == b                                        # same input -> same decision
+    # WHAT is never changed: decide reads urgency, never mutates the request (frozen).
+    assert r.side is Side.BUY and r.size == Decimal("0.001") and r.asset == "BTCUSDT"
+
+
+def test_maker_urgencies_never_taker():
+    for u in (UrgencyClass.PROTECT, UrgencyClass.HARVEST,
+              UrgencyClass.ROTATE, UrgencyClass.ACQUIRE):
+        assert decide(_req(u, deadline_ms=1000)).method is ExecutionMethod.MAKER, u
+
+
+# ------------------------------------------------------- RoutingDecision self-guard
+
+def test_bounded_ms_requires_max_ms():
+    with pytest.raises(ValueError):
+        RoutingDecision(
+            method=ExecutionMethod.MAKER, max_reprices=1,
+            ttl=TtlDiscipline.BOUNDED_MS, cross_on_expiry=True,
+            wants_placement=False, ignore_advice=False, max_ms=None,  # contradiction
+        )
+
+
+def test_non_bounded_rejects_max_ms():
+    with pytest.raises(ValueError):
+        RoutingDecision(
+            method=ExecutionMethod.MAKER, max_reprices=1,
+            ttl=TtlDiscipline.UNBOUNDED, cross_on_expiry=False,
+            wants_placement=True, ignore_advice=False, max_ms=1000,  # contradiction
+        )
+
+
+# ---------------------------------------------------------- contract validation
+
+def test_urgency_exit_classification():
+    assert UrgencyClass.ACQUIRE.is_exit is False         # entry
+    for u in (UrgencyClass.CATASTROPHIC, UrgencyClass.PROTECT,
+              UrgencyClass.HARVEST, UrgencyClass.ROTATE):
+        assert u.is_exit is True
+
+
+@pytest.mark.parametrize("bad", [
+    dict(request_id=""),                                 # missing idempotency key
+    dict(asset="BTC-USDT"),                              # dashed (dialect's job, not caller's)
+    dict(asset="btcusdt"),                               # lowercase
+    dict(size=Decimal("0")),                             # non-positive size
+    dict(size=Decimal("-1")),
+    dict(guideline_price=Decimal("0")),                  # non-positive reference
+    dict(deadline_ms=0),                                 # non-positive patience
+])
+def test_execution_request_rejects_bad_input(bad):
+    with pytest.raises(ValueError):
+        _req(UrgencyClass.HARVEST, **bad)
+
+
+def test_acquire_cannot_be_reduce_only():
+    with pytest.raises(ValueError):
+        ExecutionRequest(
+            request_id="r", asset="BTCUSDT", side=Side.BUY,
+            size=Decimal("1"), urgency=UrgencyClass.ACQUIRE, reduce_only=True,
+        )
+
+
+def test_protective_spec_needs_exactly_one_anchor():
+    with pytest.raises(ValueError):
+        ProtectiveSpec()                                  # neither
+    with pytest.raises(ValueError):
+        ProtectiveSpec(stop_price=Decimal("100"), stop_mult=Decimal("2"))  # both
+    ok = ProtectiveSpec(stop_mult=Decimal("2"))
+    assert ok.working_type == "MARK_PRICE" and ok.reduce_only is True
+
+
+def test_protective_spec_rejects_non_mark_working_type():
+    with pytest.raises(ValueError):
+        ProtectiveSpec(stop_mult=Decimal("2"), working_type="LAST_PRICE")
+
+
+@pytest.mark.parametrize("frac", [Decimal("0"), Decimal("-0.1"), Decimal("1.5")])
+def test_advice_qty_fraction_bounds(frac):
+    with pytest.raises(ValueError):
+        ExecutionAdvice(qty_fraction=frac)
+
+
+def test_advice_is_ignorable_metadata_only():
+    # Advice cannot change WHAT: it carries no side/size/asset. Structural proof.
+    adv = ExecutionAdvice(source="MALKHUT", prefer_post_only=True, note="rest deeper")
+    fields = adv.__dataclass_fields__
+    for forbidden in ("side", "size", "asset", "reduce_only"):
+        assert forbidden not in fields