鸽子的回忆录

if mode == "algo":
    action = algo_solver.choose(...)
elif mode == "llm":
    action = llm_strategy.ask(prompt=..., context=...)
elif mode == "rl":
    obs = build_observation(...)
    action = rl_policy.predict(obs)

class BaseStrategy(ABC):
    @abstractmethod
    def decide(self, state: GameState) -> Decision:
        """根据当前游戏状态做决策"""

class AlgoStrategy(BaseStrategy):
    def decide(self, state: GameState) -> Decision:
        best_action = None
        best_eval = -inf
        for action in state.legal_actions:
            ev = evaluate(state, action)
            if ev > best_eval:
                best_eval, best_action = ev, action
        return Decision(
            action=best_action,
            reason=f"evaluation={best_eval:.2f}",
            meta={"all_evals": {...}},
        )

@dataclass
class GameState:
    # 当前局面
    hand: list[Card]
    hp: int
    turn: int
    ...
    # 历史上下文（由上层维护，显式传入）
    session_history: SessionHistory
    insights: list[Insight]

class LLMStrategy(BaseStrategy):
    def decide(self, state: GameState) -> Decision:
        prompt = self._build_prompt(state)  # 完全由 state 构造
        response = self.llm_client.chat(prompt)
        return self._parse(response, state)

class RLStrategy(BaseStrategy):
    def __init__(self, ckpt_path):
        self.policy = load_policy(ckpt_path)
    
    def decide(self, state):
        obs = state_to_observation(state)
        action = self.policy.predict(obs)
        return Decision(action=action, ...)

# rl_inference_client.py
class RLInferenceClient:
    def predict(self, observation: dict) -> int:
        resp = self.http.post(self.endpoint, json=observation)
        return resp.json()["action"]

# rl_strategy.py
class RLStrategy(BaseStrategy):
    def __init__(self, client: RLInferenceClient):
        self.client = client
    
    def decide(self, state):
        obs = state_to_observation(state)
        action = self.client.predict(obs)
        return Decision(action=action, ...)

@dataclass(frozen=True)
class Decision:
    action: Action               # 必须是已知 Action 枚举
    reason: str                  # 必须有
    meta: dict[str, Any] = field(default_factory=dict)  # 各家自由发挥

def get_strategy(scene: Scene, config: Config) -> BaseStrategy:
    # 配置层面的强制指定优先
    if config.force_strategy:
        return STRATEGIES[config.force_strategy]
    
    # 按场景默认
    return SCENE_DEFAULT_STRATEGY[scene]

SCENE_DEFAULT_STRATEGY = {
    Scene.DAILY_TASK: "algo",      # 日常任务，启发式够用
    Scene.HIGH_RANK_EXAM: "rl",    # 高难度，RL 冲分
    Scene.STORY_DIALOGUE: "llm",   # 剧情对话，LLM 选项
    ...
}

def decide_with_fallback(state, primary, fallbacks=[algo, safe]):
    for strategy in [primary, *fallbacks]:
        try:
            return strategy.decide(state)
        except StrategyError as e:
            log.warning(f"{strategy} failed: {e}, trying next")
    raise RuntimeError("All strategies failed")

class YoloModelFromONNX:
    def __init__(self, model_path: str):
        # ...
        self._agnostic_nms_groups = self._build_agnostic_nms_groups()

    def _build_agnostic_nms_groups(self) -> list | None:
        skill_labels = {"Skill Card: Active", "Skill Card: Mental", "Skill Card: Trap"}
        group = set()

        for cid, name in self._model_meta.names.items():
            if name in skill_labels:
                group.add(cid)

        return [group] if len(group) >= 2 else None

import cv2
import numpy as np

def _postprocess(
    self,
    results: np.ndarray,
    conf_threshold: float,
    iou_threshold: float,
    agnostic_nms_groups: list | None = None,
) -> YoloResult:
    boxes = []
    scores = []
    class_ids = []

    for detection in results:
        class_scores = detection[5:]
        max_score = np.amax(class_scores)
        if max_score >= conf_threshold:
            class_id = np.argmax(class_scores)
            x, y, w, h = detection[:4]
            boxes.append([x - w/2, y - h/2, w, h])
            scores.append(max_score)
            class_ids.append(class_id)

    if not boxes:
        return YoloResult.empty()

    # 类别 → 组索引
    agnostic_map = {}
    if agnostic_nms_groups:
        for gi, group in enumerate(agnostic_nms_groups):
            for cid in group:
                agnostic_map[cid] = gi

    # 分桶
    grouped = {}
    for i, cid in enumerate(class_ids):
        if cid in agnostic_map:
            key = ("agnostic", agnostic_map[cid])
        else:
            key = ("class", int(cid))
        grouped.setdefault(key, []).append(i)

    # 每个桶独立 NMS
    final_indices = []
    for indices_in_group in grouped.values():
        group_boxes = [boxes[i] for i in indices_in_group]
        group_scores = [scores[i] for i in indices_in_group]

        keep = cv2.dnn.NMSBoxes(
            group_boxes, group_scores, conf_threshold, iou_threshold
        )

        if keep is not None:
            keep = keep.flatten()
            for idx in keep:
                final_indices.append(indices_in_group[idx])

    return YoloResult(
        boxes=np.array([boxes[i] for i in final_indices]),
        scores=np.array([scores[i] for i in final_indices]),
        class_ids=np.array([class_ids[i] for i in final_indices]),
    )

skill_labels = {"Skill Card: Active", "Skill Card: Mental", "Skill Card: Trap"}
agnostic_nms_groups = [skill_labels]

button_labels = {"Button: Normal", "Button: Pressed", "Button: Disabled"}
agnostic_nms_groups = [button_labels]

agnostic_nms_groups = [
    {"Skill Card: Active", "Skill Card: Mental", "Skill Card: Trap"},
    {"Button: Normal", "Button: Pressed"},
]

import numpy as np
import onnxruntime as ort
import cv2

class CLIPModelFromONNX:
    def __init__(self, model_path: str):
        self.session = ort.InferenceSession(model_path)
        self._input_name = self.session.get_inputs()[0].name

    def _preprocess(self, image: np.ndarray) -> np.ndarray:
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        image, _, _ = letterbox(image, (224, 224))
        image = center_crop(image)
        image = image.astype(np.float32) / 255.0
        mean = np.array([0.48145466, 0.4578275, 0.40821073])
        std = np.array([0.26862954, 0.26130258, 0.27577711])
        image = (image - mean) / std
        image = np.transpose(image, (2, 0, 1))
        return image[np.newaxis, :].astype(np.float32)

    def forward(self, image: np.ndarray) -> np.ndarray:
        input_tensor = self._preprocess(image)
        output = self.session.run(None, {self._input_name: input_tensor})
        return output[0]

import pickle
from dataclasses import dataclass

@dataclass
class CLIPRetrieveData:
    payload: Any
    similarity: float

class CLIPMemory:
    def __init__(self, clip_name: str):
        self._clip_name = clip_name
        self._engine = CLIPModelFromONNX("clip_visual.onnx")
        self._image_file_path = f"data/CLIP/{clip_name}"

    @staticmethod
    def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
        a, b = a.flatten(), b.flatten()
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def add_to_memory(self, image, payload, threshold=0.96) -> bool:
        features = self._engine.forward(image)
        for memory in self._load_memories():
            existing = pickle.loads(memory.features)
            if self._cosine_similarity(features, existing) > threshold:
                return False
        self._save_memory(features, payload)
        return True

    def retrieve(self, image, threshold=0.96):
        features = self._engine.forward(image)
        best_sim, best_payload = -1, None
        for memory in self._load_memories():
            existing = pickle.loads(memory.features)
            sim = self._cosine_similarity(features, existing)
            if sim > best_sim:
                best_sim, best_payload = sim, memory.payload

        if best_payload is not None and best_sim > threshold:
            return CLIPRetrieveData(best_payload, best_sim)
        return None

def augment_image(image: np.ndarray) -> list:
    augmented = []
    h, w = image.shape[:2]

    # JPEG 压缩
    _, buf = cv2.imencode(".jpg", image, [cv2.IMWRITE_JPEG_QUALITY, 30])
    augmented.append(cv2.imdecode(buf, cv2.IMREAD_COLOR))

    # 高斯噪点
    noise = np.random.normal(0, 15, image.shape).astype(np.float32)
    noisy = np.clip(image.astype(np.float32) + noise, 0, 255).astype(np.uint8)
    augmented.append(noisy)

    # 亮度偏移
    augmented.append(cv2.convertScaleAbs(image, alpha=1.15, beta=20))

    # 色调偏移
    hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV).astype(np.int16)
    hsv[:, :, 0] = (hsv[:, :, 0] + 8) % 180
    augmented.append(cv2.cvtColor(hsv.astype(np.uint8), cv2.COLOR_HSV2BGR))

    # 缩放
    small = cv2.resize(image, (int(w * 0.85), int(h * 0.85)))
    augmented.append(cv2.resize(small, (w, h)))

    return augmented

def add_to_memory_with_augmentation(self, image, payload):
    self.add_to_memory(image, payload)
    orig_features = self._engine.forward(image)
    for aug_image in augment_image(image):
        aug_features = self._engine.forward(aug_image)
        if self._cosine_similarity(orig_features, aug_features) < 0.95:
            self._save_memory(aug_features, payload)

def validate_inclass_similarity(self, image, target_id) -> bool:
    MIN_INCLASS_SIMILARITY = 0.72
    image_features = self._engine.forward(image)
    best_inclass = -1.0
    for memory in self._load_memories():
        if memory.payload_id != target_id:
            continue
        existing = pickle.loads(memory.features)
        sim = self._cosine_similarity(image_features, existing)
        best_inclass = max(best_inclass, sim)

    if best_inclass >= 0 and best_inclass < MIN_INCLASS_SIMILARITY:
        return False
    return True

def retrieve_with_cleanup(self, image, threshold=0.96):
    features = self._engine.forward(image)
    best_sim, best_payload = -1, None
    stale_ids = []

    for memory in self._load_memories():
        try:
            payload = memory.load_payload()
            if payload is None:
                stale_ids.append(memory.id)
                continue
        except Exception:
            stale_ids.append(memory.id)
            continue

        existing = pickle.loads(memory.features)
        sim = self._cosine_similarity(features, existing)
        if sim > best_sim:
            best_sim, best_payload = sim, payload

    if stale_ids:
        self._delete_memories(stale_ids)

    if best_payload and best_sim > threshold:
        return CLIPRetrieveData(best_payload, best_sim)
    return None

class CLIPServiceManager:
    def __init__(self):
        self._model_session = CLIPModelFromONNX("clip_visual.onnx")
        self.support_card_clip = SupportCardCLIP(self._model_session)
        self.skill_card_clip = SkillCardCLIP(self._model_session)
        self.item_clip = ItemCLIP(self._model_session)
        self.idol_card_clip = IdolCardCLIP(self._model_session)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List

@dataclass
class OCRResult:
    text: str
    confidence: float
    bbox: tuple  # (x1, y1, x2, y2)

@dataclass
class OCRResponse:
    results: List[OCRResult]
    backend: str

class BaseOCRBackend(ABC):
    @abstractmethod
    def ocr(self, image) -> OCRResponse: ...

    @abstractmethod
    def is_available(self) -> bool: ...

class RapidOCRBackend(BaseOCRBackend):
    def __init__(self):
        from rapidocr_onnxruntime import RapidOCR
        self._engine = RapidOCR()

    def ocr(self, image) -> OCRResponse:
        result, _ = self._engine(image)
        if result is None:
            return OCRResponse(results=[], backend="rapidocr")

        ocr_results = []
        for line in result:
            bbox, text, confidence = line
            ocr_results.append(OCRResult(text=text, confidence=confidence, bbox=bbox))
        return OCRResponse(results=ocr_results, backend="rapidocr")

    def is_available(self) -> bool:
        try:
            import rapidocr_onnxruntime
            return True
        except ImportError:
            return False

class VisionOCRBackend(BaseOCRBackend):
    def ocr(self, image) -> OCRResponse:
        import Vision
        import Quartz

        ci_image = Quartz.CIImage.imageWithCGImage_(image)

        request = Vision.VNRecognizeTextRequest.alloc().init()
        request.setRecognitionLanguages_(["zh-Hans", "en"])
        request.setRecognitionLevel_(Vision.VNRequestTextRecognitionLevelAccurate)

        handler = Vision.VNImageRequestHandler.alloc().initWithCIImage_options_(
            ci_image, None
        )
        success = handler.performRequests_error_([request], None)

        if not success:
            return OCRResponse(results=[], backend="vision")

        ocr_results = []
        for observation in request.results():
            text = observation.topCandidates_(1)[0].string()
            confidence = observation.topCandidates_(1)[0].confidence()
            bbox = observation.boundingBox()
            ocr_results.append(OCRResult(text=text, confidence=confidence, bbox=bbox))
        return OCRResponse(results=ocr_results, backend="vision")

    def is_available(self) -> bool:
        import platform
        return platform.system() == "Darwin"

from enum import Enum

class OCRBackendType(str, Enum):
    AUTO = "auto"
    RAPIDOCR = "rapidocr"
    VISION = "vision"

def create_ocr_backend(requested_backend: str = None) -> BaseOCRBackend:
    requested = requested_backend or get_configured_backend()
    candidates = resolve_candidates(requested)

    errors = []
    for candidate in candidates:
        try:
            backend = _build_backend(candidate)
            if requested == "auto":
                logger.info(f"Using OCR backend {candidate} (auto)")
            elif candidate != requested:
                logger.warning(f"OCR backend {requested} unavailable, fallback to {candidate}")
            return backend
        except Exception as exc:
            errors.append(f"{candidate}: {exc}")
            logger.warning(f"Failed to initialize OCR backend {candidate}: {exc}")

    raise RuntimeError(f"Failed to initialize any OCR backend: {' | '.join(errors)}")

def _build_backend(backend_name: str) -> BaseOCRBackend:
    if backend_name == "vision":
        return VisionOCRBackend()
    if backend_name == "rapidocr":
        return RapidOCRBackend()
    raise ValueError(f"Unknown OCR backend: {backend_name}")

import platform

def resolve_candidates(requested: str) -> list:
    current_platform = platform.system()

    if requested == "auto":
        if current_platform == "Darwin":
            return ["vision", "rapidocr"]  # macOS：Vision 优先
        return ["rapidocr"]

    if requested == "vision":
        # 用户指名要 Vision，但失败了就降级到 RapidOCR
        return ["vision", "rapidocr"]

    return ["rapidocr"]

import os

def get_configured_backend() -> str:
    if env_backend := os.getenv("GAKUMAS_OCR_BACKEND"):
        return normalize_backend_name(env_backend)

    from config import ConfigService
    return ConfigService().base.ocr_backend

def normalize_backend_name(value: str) -> str:
    normalized = value.strip().lower()
    valid_values = ["auto", "rapidocr", "vision"]
    if normalized in valid_values:
        return normalized
    return "auto"

class OCRService:
    def __init__(self):
        self._backend = create_ocr_backend()

    def ocr(self, image) -> OCRResponse:
        return self._backend.ocr(image)

    def ocr_text(self, image) -> str:
        response = self.ocr(image)
        return " ".join(r.text for r in response.results)

import threading
from time import sleep
from typing import Callable, List

class YoloInferenceEngine:
    def __init__(self, device):
        self._device = device
        self._engine = None
        self._latest_results = None

        self._infer_callback_list: List[Callable] = []
        self._capture_failure_callback_list: List[Callable] = []

        self.__flag_loop = False
        self.__flag_pause = False

        self.__action_lock = threading.Lock()
        self.__result_write_lock = threading.Lock()

    def start(self):
        with self.__action_lock:
            if self.__flag_loop:
                return False
            self.__flag_pause = False
            self.__flag_loop = True
            self._capture_thread = threading.Thread(
                target=self._inference_loop,
                daemon=True,
            )
            self._capture_thread.start()
        return True

    def stop(self):
        with self.__action_lock:
            if not self.__flag_loop:
                return False
            self.__flag_pause = False
            self.__flag_loop = False
            self._capture_thread.join(timeout=3)
        return True

    def pause(self):
        with self.__action_lock:
            if self.__flag_pause:
                return False
            self.__flag_pause = True
            return True

    def resume(self):
        with self.__action_lock:
            if not self.__flag_pause:
                return False
            self.__flag_pause = False
            return True

def _inference_loop(self):
    while self.__flag_loop:
        if self.__flag_pause:
            sleep(0.1)
            continue

        try:
            frame = self._device.capture()
        except Exception as e:
            self._exec_capture_failure_callback(e)
            self.__flag_loop = False
            return

        if frame is None or frame.size <= 0:
            sleep(0.1)
            continue

        results = self._engine(
            frame,
            conf_threshold=0.6,
            agnostic_nms_groups=self._agnostic_nms_groups,
        )

        with self.__result_write_lock:
            self._latest_results = Yolo_Results(results, frame)

        self._exec_infer_callback()

    self.__flag_loop = False

def register_infer_callback(self, func: Callable):
    with self.__action_lock:
        if func not in self._infer_callback_list:
            self._infer_callback_list.append(func)

def register_capture_failure_callback(self, func: Callable):
    with self.__action_lock:
        if func not in self._capture_failure_callback_list:
            self._capture_failure_callback_list.append(func)

def _exec_infer_callback(self):
    for callback in self._infer_callback_list:
        try:
            callback(self.latest_frame, self.latest_results)
        except Exception as e:
            logger.error(f"推理回调失败: {e}")

def _exec_capture_failure_callback(self, exc: Exception):
    for callback in self._capture_failure_callback_list:
        try:
            callback(exc)
        except Exception as callback_exc:
            logger.error(f"截图失败回调失败: {callback_exc}")

engine = YoloInferenceEngine(device)

def on_inference_result(frame, results):
    for box, score, class_id in zip(results.boxes, results.scores, results.class_ids):
        if score > 0.8:
            print(f"检测到: {class_id}, 置信度: {score:.2f}")

engine.register_infer_callback(on_inference_result)
engine.start()

@property
def running(self) -> bool:
    with self.__action_lock:
        return self.__flag_loop

@property
def is_pause(self) -> bool:
    with self.__action_lock:
        return self.__flag_pause

@property
def latest_frame(self):
    try:
        return self._latest_results.frame
    except Exception:
        return None

@property
def latest_results(self):
    try:
        return self._latest_results
    except Exception:
        return None

def load_model(self, model_type: str = YoloModelType.BASE_UI):
    if self.__flag_loop:
        self.pause()

    with self.__action_lock:
        self._engine = YoloModelFromONNX(config.model_config[model_type])
        self._model_type = model_type
        self._agnostic_nms_groups = self._build_skill_card_nms_group()

    if self.__flag_loop:
        self.resume()

import onnxruntime as ort

PREFERRED_EXECUTION_PROVIDERS = (
    "DmlExecutionProvider",      # Windows
    "CoreMLExecutionProvider",   # macOS
    "CPUExecutionProvider",      # 兜底
)

class DMLManager:
    @classmethod
    def get_session_providers(cls) -> list[str]:
        available_providers = set(ort.get_available_providers())
        providers = [
            provider
            for provider in cls._preferred_execution_providers
            if provider in available_providers
        ]
        if "CPUExecutionProvider" not in providers:
            providers.append("CPUExecutionProvider")
        return providers

    @classmethod
    def create_session(cls, model_path: str) -> ort.InferenceSession:
        providers = cls.get_session_providers()
        so = cls._build_session_options()
        try:
            return ort.InferenceSession(model_path, sess_options=so, providers=providers)
        except Exception as exc:
            logger.warning(f"加速后端失败，回退到 CPU: {exc}")
            return ort.InferenceSession(
                model_path,
                sess_options=so,
                providers=["CPUExecutionProvider"],
            )

import hashlib
from pathlib import Path

class DMLManager:
    @classmethod
    def _build_model_cache_key(cls, model_path: str, provider_options: dict) -> str:
        model_file = Path(model_path).resolve()
        digest = hashlib.sha256()

        # 模型本体和 .data 外部权重都要算进去
        for file_path in cls._iter_model_fingerprint_files(model_file):
            with open(file_path, "rb") as f:
                for chunk in iter(lambda: f.read(1024 * 1024), b""):
                    digest.update(chunk)

        digest.update(repr(sorted(provider_options.items())).encode())

        if cls._free_dimension_overrides:
            digest.update(repr(cls._free_dimension_overrides).encode())

        return digest.hexdigest()[:24]

    @classmethod
    def _build_provider_config(cls, model_path: str) -> list:
        providers = []
        available = cls.get_session_providers()

        if "DmlExecutionProvider" in available:
            providers.append("DmlExecutionProvider")

        if "CoreMLExecutionProvider" in available:
            cache_root = cls._get_cache_root()
            cache_dir = cache_root / "coreml-cache"

            provider_options = {
                "ModelFormat": "MLProgram",
                "MLComputeUnits": "ALL",
                "RequireStaticInputShapes": "0",
            }
            model_cache_key = cls._build_model_cache_key(model_path, provider_options)
            cache_subdir = cache_dir / model_cache_key
            cache_subdir.mkdir(parents=True, exist_ok=True)

            # 顺手清旧版本的缓存
            cls._prune_stale_coreml_cache(cache_dir, model_cache_key)

            providers.append((
                "CoreMLExecutionProvider",
                {**provider_options, "ModelCacheDirectory": str(cache_subdir)},
            ))

        if "CPUExecutionProvider" in available:
            providers.append("CPUExecutionProvider")

        return providers

@classmethod
def create_session_with_retry(cls, model_path: str) -> ort.InferenceSession:
    providers = cls._build_provider_config(model_path)
    so = cls._build_session_options()

    try:
        return ort.InferenceSession(model_path, sess_options=so, providers=providers)
    except Exception as exc:
        coreml_cache_dir = cls._extract_coreml_cache_dir(providers)
        if coreml_cache_dir is not None:
            cls._clear_coreml_cache_dir(coreml_cache_dir)
            logger.warning(f"清除 CoreML 缓存后重试: {exc}")
            providers = cls._build_provider_config(model_path)
            try:
                return ort.InferenceSession(model_path, sess_options=so, providers=providers)
            except Exception as retry_exc:
                logger.warning(f"重试失败，回退到 CPU: {retry_exc}")

        return ort.InferenceSession(
            model_path,
            sess_options=so,
            providers=["CPUExecutionProvider"],
        )

class DMLManager:
    _free_dimension_overrides = [
        ("batch", 1),
        ("batch_size", 1),
    ]

    @classmethod
    def _build_session_options(cls, extra_overrides: dict = None) -> ort.SessionOptions:
        so = ort.SessionOptions()
        so.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL
        so.intra_op_num_threads = 1
        so.inter_op_num_threads = 1

        for name, value in cls._free_dimension_overrides:
            so.add_free_dimension_override_by_name(name, value)

        if extra_overrides:
            for name, value in extra_overrides.items():
                so.add_free_dimension_override_by_name(name, value)
        return so

    @classmethod
    def _build_dim_overrides_for_model(cls, model_path: str) -> dict:
        overrides = {}
        imgsz = cls._read_model_imgsz(model_path)
        if imgsz is not None:
            overrides["height"] = imgsz[0]
            overrides["width"] = imgsz[1]
        return overrides

class DMLManager:
    _session_replacements: dict[int, ort.InferenceSession] = {}

    @classmethod
    def run(cls, session: ort.InferenceSession, feeds: dict) -> list:
        with cls._lock:
            active_session = cls._resolve_session(session)
            try:
                return active_session.run(None, feeds)
            except Exception as exc:
                fallback_session = cls._fallback_to_cpu_session(active_session, exc)
                if fallback_session is None:
                    raise
                return fallback_session.run(None, feeds)

    @classmethod
    def _fallback_to_cpu_session(cls, session, exc) -> ort.InferenceSession | None:
        session_id = id(session)
        model_path = cls._session_model_paths.get(session_id)
        provider_names = cls._session_provider_names.get(session_id)

        if not model_path or not provider_names:
            return None
        if tuple(provider_names) == ("CPUExecutionProvider",):
            return None  # 已经是 CPU 了

        logger.warning(f"ONNX 推理失败，回退到 CPU: {exc}")

        fallback_session = cls._create_cpu_session(model_path)
        # 同一会话以后再来，直接走 CPU
        cls._session_replacements[session_id] = fallback_session
        return fallback_session

    @classmethod
    def _resolve_session(cls, session) -> ort.InferenceSession:
        current = session
        visited = set()
        while True:
            current_id = id(current)
            replacement = cls._session_replacements.get(current_id)
            if replacement is None or current_id in visited:
                return current
            visited.add(current_id)
            current = replacement