Initialize all v2 subsystems. Call once at proxy startup.
(data_dir: Path | None = None)
| 48 | |
| 49 | |
| 50 | def on_startup(data_dir: Path | None = None) -> None: |
| 51 | """Initialize all v2 subsystems. Call once at proxy startup.""" |
| 52 | global _metrics, _shadow, _weight_tracker, _index_manager, _state_dir, _initialized |
| 53 | |
| 54 | if _initialized: |
| 55 | return |
| 56 | _initialized = True |
| 57 | |
| 58 | if data_dir is None: |
| 59 | from uncommon_route.paths import data_dir as get_data_dir |
| 60 | data_dir = get_data_dir() |
| 61 | _state_dir = data_dir |
| 62 | |
| 63 | # Load persisted state |
| 64 | state = load_state(data_dir) |
| 65 | logger.info( |
| 66 | "v2 lifecycle started: weights=%s, shadow_promoted=%s, shadow_streak=%d", |
| 67 | state.signal_weights, state.shadow_promoted, state.shadow_consecutive_wins, |
| 68 | ) |
| 69 | |
| 70 | # Initialize subsystems from persisted state |
| 71 | _metrics = RoutingMetrics() |
| 72 | _weight_tracker = SignalWeightTracker(initial_weights=state.signal_weights) |
| 73 | _shadow = ShadowTracker(eval_window=200, promote_after=3) |
| 74 | _shadow._consecutive_wins = state.shadow_consecutive_wins |
| 75 | _shadow._promoted = state.shadow_promoted |
| 76 | |
| 77 | # Embedding index manager |
| 78 | try: |
| 79 | splits_dir = data_dir / "v2_splits" |
| 80 | emb_path = splits_dir / "seed_embeddings.npy" |
| 81 | labels_path = splits_dir / "seed_labels.json" |
| 82 | if emb_path.exists() and labels_path.exists(): |
| 83 | _index_manager = EmbeddingIndexManager( |
| 84 | index_path=emb_path, labels_path=labels_path, |
| 85 | max_size=10_000, dedup_threshold=0.95, |
| 86 | ) |
| 87 | logger.info("v2 embedding index manager loaded (%d entries)", _index_manager.size) |
| 88 | except Exception as e: |
| 89 | logger.warning("v2 embedding index manager init failed: %s", e) |
| 90 | |
| 91 | |
| 92 | def on_shutdown() -> None: |
nothing calls this directly
no test coverage detected