refactor(b09): drop bound-cycle STRATEGY_RUNTIME module fallback

main.py

@@ -268,8 +268,10 @@ def _load_import_safe_strategy_runtime():
 def _activate_execution_strategy_runtime(profile: str, *, runtime_target=None):
     """Replace the import-safe evaluator only after RuntimeTarget validation.
 
-    Writes the activated handle to module STRATEGY_RUNTIME (compat / unbound
-    fallback) and, when a RunContext is bound, also to ctx.strategy_runtime.
+    When a RunContext is bound (cycle path), mount the handle on the context only
+    — do not rewrite module STRATEGY_RUNTIME (no cycle→module leakage). Unbound
+    callers still update the module symbol for out-of-cycle compat / patches.
+    Pool-derived knobs (size / artifact paths) still refresh either way.
     """
 
     global STRATEGY_RUNTIME
@@ -283,14 +285,15 @@ def _activate_execution_strategy_runtime(profile: str, *, runtime_target=None):
         if runtime_target is None
         else load_strategy_runtime(profile, runtime_target=runtime_target)
     )
-    STRATEGY_RUNTIME = activated
     TREND_POOL_SIZE = activated.trend_pool_size
     DEFAULT_LIVE_POOL_LEGACY_PATH = activated.default_local_artifact_path
     DEFAULT_TREND_POOL_MAX_AGE_DAYS = int(activated.artifact_contract["max_age_days"])
     DEFAULT_TREND_POOL_ACCEPTABLE_MODES = tuple(activated.artifact_contract["acceptable_modes"])
     ctx = get_active_run_context()
     if ctx is not None:
         ctx.strategy_runtime = activated
+        return activated
+    STRATEGY_RUNTIME = activated
     return activated
 
 
@@ -1350,9 +1353,12 @@ def _execute_btc_dca_cycle(
 
 
 def execute_cycle(runtime):
-    # B09: cycle-scoped universe + strategy handle on RunContext.
-    # Do not mutate module TREND_UNIVERSE; STRATEGY_RUNTIME stays as unbound fallback.
+    # B09 PR-3: cycle-scoped universe + strategy handle on RunContext.
+    # Bind-time seed may copy module defaults once; while bound, resolves never
+    # fall back to module STRATEGY_RUNTIME / TREND_UNIVERSE.
     mounted_strategy = getattr(runtime, "strategy_runtime", None) or STRATEGY_RUNTIME
+    if mounted_strategy is None:
+        raise RuntimeError("execute_cycle_missing_strategy_runtime")
     ctx = RunContext(
         trend_universe={symbol: meta.copy() for symbol, meta in TREND_UNIVERSE.items()},
         strategy_runtime=mounted_strategy,

run_context.py

@@ -1,8 +1,9 @@
 """Per-cycle RunContext for BinancePlatform (B09).
 
-Trend universe (PR-1) and strategy-runtime handle (PR-2) bind to a cycle-scoped
-context during execute_cycle. Module STRATEGY_RUNTIME / TREND_UNIVERSE remain
-import-safe fallbacks and out-of-cycle patch points.
+Trend universe and strategy-runtime bind to a cycle-scoped context during
+execute_cycle. While a RunContext is bound, resolves do not fall back to module
+STRATEGY_RUNTIME / TREND_UNIVERSE. Module symbols remain import-safe seeds and
+out-of-cycle patch points only.
 Does not change order-submission or state-owner semantics in runtime_support.
 """
 
@@ -27,6 +28,10 @@ class RunContext:
 )
 
 
+class RunContextError(RuntimeError):
+    """Fail-closed when a bound cycle is missing a required RunContext field."""
+
+
 def bind_run_context(ctx: RunContext) -> Token:
     return _ACTIVE_RUN_CONTEXT.set(ctx)
 
@@ -40,11 +45,13 @@ def get_active_run_context() -> Optional[RunContext]:
 
 
 def resolve_trend_universe(
-    fallback: Mapping[str, dict[str, Any]],
+    fallback: Mapping[str, dict[str, Any]] | None = None,
 ) -> Mapping[str, dict[str, Any]]:
     ctx = _ACTIVE_RUN_CONTEXT.get()
     if ctx is not None:
         return ctx.trend_universe
+    if fallback is None:
+        raise RunContextError("trend_universe_unbound_requires_fallback")
     return fallback
 
 
@@ -60,11 +67,15 @@ def set_active_trend_universe(
     fallback_setter(resolved)
 
 
-def resolve_strategy_runtime(fallback: Any) -> Any:
-    """Prefer cycle-bound strategy runtime; else module/import-safe fallback."""
+def resolve_strategy_runtime(fallback: Any = None) -> Any:
+    """Bound cycles must use ctx.strategy_runtime; unbound may use module fallback."""
     ctx = _ACTIVE_RUN_CONTEXT.get()
-    if ctx is not None and ctx.strategy_runtime is not None:
+    if ctx is not None:
+        if ctx.strategy_runtime is None:
+            raise RunContextError("run_context_missing_strategy_runtime")
         return ctx.strategy_runtime
+    if fallback is None:
+        raise RunContextError("strategy_runtime_unbound_requires_fallback")
     return fallback
 
 

tests/test_run_context.py

@@ -9,6 +9,7 @@
 import main
 from run_context import (
     RunContext,
+    RunContextError,
     bind_run_context,
     get_active_run_context,
     reset_run_context,
@@ -38,6 +39,12 @@ def test_resolve_falls_back_to_module_when_unbound(self):
         self.assertIs(resolve_trend_universe(main.TREND_UNIVERSE), main.TREND_UNIVERSE)
         self.assertIs(resolve_strategy_runtime(main.STRATEGY_RUNTIME), main.STRATEGY_RUNTIME)
 
+    def test_unbound_resolve_without_fallback_fails_closed(self):
+        with self.assertRaises(RunContextError):
+            resolve_strategy_runtime()
+        with self.assertRaises(RunContextError):
+            resolve_trend_universe()
+
     def test_set_and_resolve_use_active_context(self):
         custom = {"ZZZTESTUSDT": {"base_asset": "ZZZ"}}
         ctx = RunContext(trend_universe=_baseline_universe())
@@ -81,18 +88,33 @@ def test_sequential_binds_do_not_leak_universe(self):
 
         self.assertEqual(dict(main.TREND_UNIVERSE), _baseline_universe())
 
-    def test_resolve_strategy_runtime_prefers_active_context(self):
+    def test_bound_resolve_ignores_poisoned_module_strategy_runtime(self):
+        """PR-3: while bound, module STRATEGY_RUNTIME must not be consulted."""
         module_handle = main.STRATEGY_RUNTIME
         bound_handle = SimpleNamespace(profile="bound-profile", marker="ctx")
+        poison = SimpleNamespace(profile="poison-module", marker="module")
         ctx = RunContext(trend_universe=_baseline_universe(), strategy_runtime=bound_handle)
         token = bind_run_context(ctx)
         try:
+            main.STRATEGY_RUNTIME = poison
+            # Fallback arg is also poisoned — bound path must ignore both.
+            self.assertIs(resolve_strategy_runtime(poison), bound_handle)
             self.assertIs(resolve_strategy_runtime(module_handle), bound_handle)
-            self.assertIs(main.STRATEGY_RUNTIME, module_handle)
+            self.assertIs(main.STRATEGY_RUNTIME, poison)
         finally:
             reset_run_context(token)
+            main.STRATEGY_RUNTIME = module_handle
         self.assertIs(resolve_strategy_runtime(module_handle), module_handle)
 
+    def test_bound_without_strategy_runtime_fails_closed(self):
+        ctx = RunContext(trend_universe=_baseline_universe(), strategy_runtime=None)
+        token = bind_run_context(ctx)
+        try:
+            with self.assertRaises(RunContextError):
+                resolve_strategy_runtime(main.STRATEGY_RUNTIME)
+        finally:
+            reset_run_context(token)
+
     def test_sequential_binds_do_not_leak_strategy_runtime(self):
         first = SimpleNamespace(profile="first")
         second = SimpleNamespace(profile="second")
@@ -183,7 +205,36 @@ def fake_cycle(runtime, **_kwargs):
         self.assertEqual(report["status"], "ok")
         self.assertIsNone(get_active_run_context())
 
-    def test_activate_while_bound_writes_context_and_module(self):
+    def test_execute_cycle_ignores_module_strategy_poison_and_does_not_leak(self):
+        """PR-3: cycle resolve ignores module poison; module STRATEGY_RUNTIME unchanged."""
+        prior = main.STRATEGY_RUNTIME
+        mounted = SimpleNamespace(profile="cycle-mounted", marker="rt")
+        poison = SimpleNamespace(profile="poison", marker="module")
+
+        def fake_cycle(runtime, **_kwargs):
+            main.STRATEGY_RUNTIME = poison
+            self.assertIs(resolve_strategy_runtime(poison), mounted)
+            self.assertIs(runtime.run_context.strategy_runtime, mounted)
+            return {"status": "ok"}
+
+        runtime = ExecutionRuntime(
+            dry_run=True,
+            strategy_profile="crypto_live_pool_rotation",
+            strategy_runtime=mounted,
+        )
+        try:
+            with patch("main.execute_strategy_cycle", side_effect=fake_cycle):
+                report = main.execute_cycle(runtime)
+            self.assertEqual(report["status"], "ok")
+            # Poison applied inside cycle must not be "restored over" — we prove
+            # cycle itself did not write module STRATEGY_RUNTIME; caller poison remains.
+            self.assertIs(main.STRATEGY_RUNTIME, poison)
+        finally:
+            main.STRATEGY_RUNTIME = prior
+        self.assertIsNone(get_active_run_context())
+
+    def test_activate_while_bound_writes_context_not_module(self):
+        """PR-3: activate during a bound cycle must not leak into module STRATEGY_RUNTIME."""
         prior = main.STRATEGY_RUNTIME
         prior_pool = main.TREND_POOL_SIZE
         prior_legacy = main.DEFAULT_LIVE_POOL_LEGACY_PATH
@@ -199,15 +250,18 @@ def test_activate_while_bound_writes_context_and_module(self):
             },
             local_artifact_candidates=(),
         )
-        ctx = RunContext(trend_universe=_baseline_universe())
+        ctx = RunContext(
+            trend_universe=_baseline_universe(),
+            strategy_runtime=SimpleNamespace(profile="seed"),
+        )
         token = bind_run_context(ctx)
         try:
             with patch("main.load_strategy_runtime", return_value=fake) as loader:
                 activated = main._activate_execution_strategy_runtime("crypto_live_pool_rotation")
             loader.assert_called_once_with("crypto_live_pool_rotation")
             self.assertIs(activated, fake)
             self.assertIs(ctx.strategy_runtime, fake)
-            self.assertIs(main.STRATEGY_RUNTIME, fake)
+            self.assertIs(main.STRATEGY_RUNTIME, prior)  # no module leakage
             self.assertIs(resolve_strategy_runtime(prior), fake)
             self.assertEqual(main.TREND_POOL_SIZE, 11)
         finally:
@@ -218,16 +272,44 @@ def test_activate_while_bound_writes_context_and_module(self):
             main.DEFAULT_TREND_POOL_MAX_AGE_DAYS = prior_max_age
             main.DEFAULT_TREND_POOL_ACCEPTABLE_MODES = prior_modes
 
+    def test_activate_unbound_still_updates_module_for_compat(self):
+        prior = main.STRATEGY_RUNTIME
+        prior_pool = main.TREND_POOL_SIZE
+        prior_legacy = main.DEFAULT_LIVE_POOL_LEGACY_PATH
+        prior_max_age = main.DEFAULT_TREND_POOL_MAX_AGE_DAYS
+        prior_modes = main.DEFAULT_TREND_POOL_ACCEPTABLE_MODES
+        fake = SimpleNamespace(
+            profile="unbound-activate",
+            trend_pool_size=7,
+            default_local_artifact_path=prior.default_local_artifact_path,
+            artifact_contract={
+                "max_age_days": int(prior.artifact_contract["max_age_days"]),
+                "acceptable_modes": tuple(prior.artifact_contract["acceptable_modes"]),
+            },
+            local_artifact_candidates=(),
+        )
+        self.assertIsNone(get_active_run_context())
+        try:
+            with patch("main.load_strategy_runtime", return_value=fake):
+                activated = main._activate_execution_strategy_runtime("crypto_live_pool_rotation")
+            self.assertIs(activated, fake)
+            self.assertIs(main.STRATEGY_RUNTIME, fake)
+            self.assertEqual(main.TREND_POOL_SIZE, 7)
+        finally:
+            main.STRATEGY_RUNTIME = prior
+            main.TREND_POOL_SIZE = prior_pool
+            main.DEFAULT_LIVE_POOL_LEGACY_PATH = prior_legacy
+            main.DEFAULT_TREND_POOL_MAX_AGE_DAYS = prior_max_age
+            main.DEFAULT_TREND_POOL_ACCEPTABLE_MODES = prior_modes
+
 
 class StrategyRuntimeEquivalenceTests(unittest.TestCase):
-    """Bound resolve path must match direct module handle for a frozen profile."""
+    """Bound resolve path must match direct handle for a frozen profile."""
 
     def tearDown(self):
         self.assertIsNone(get_active_run_context())
 
     def test_bound_resolve_matches_module_metrics_for_same_handle(self):
-        # Use the import-safe module handle (already loaded) — behavior-equivalent
-        # mount onto RunContext without requiring full CryptoStrategies catalog.
         activated = main.STRATEGY_RUNTIME
         universe = {
             "ETHUSDT": {"base_asset": "ETH"},