(self)
| 256 | pass |
| 257 | |
| 258 | def _load_buffer(self) -> None: |
| 259 | if not self._buffer_path: |
| 260 | return |
| 261 | try: |
| 262 | if self._buffer_path.exists(): |
| 263 | data = json.loads(self._buffer_path.read_text()) |
| 264 | for rid, entry in data.items(): |
| 265 | mode = entry.get("mode") |
| 266 | if not mode: |
| 267 | continue |
| 268 | self._buffer[rid] = RequestContext( |
| 269 | features=entry["features"], |
| 270 | tier=_normalize_tier(entry["tier"]), |
| 271 | timestamp=entry["timestamp"], |
| 272 | model=entry.get("model", ""), |
| 273 | mode=str(mode), |
| 274 | ) |
| 275 | except Exception: |
| 276 | pass |
| 277 | |
| 278 | |
| 279 | def _adjust_tier(current: str, signal: FeedbackSignal) -> str: |
no test coverage detected